
/*-------------------------------------------------------------------------
 * worker.c
 *	   PostgreSQL 逻辑复制工作进程 (应用)
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/worker.c
 *
 * NOTES
 *	  此文件包含了从远程逻辑复制流接收的逻辑更改并应用的工作进程。
 *
 *	  主工作进程 (应用) 是由逻辑复制工作进程启动器为数据库中每个启用的订阅启动的。它使用
 *	  walsender 协议与发布者通信。
 *
 *	  本模块包含面向服务器的代码，并与 walreceiver 共享 libpqwalreceiver
 *	  模块，以提供特定于 libpq 的功能。
 *
 *
 * STREAMED TRANSACTIONS
 * ---------------------
 * 流式事务（超出上游内存限制的大事务）不会立即应用，而是数据会被写入临时文件，
 * 当最终提交到达时，数据会一次性应用。
 *
 * 与常规（非流式）情况不同，处理流式事务必须处理顶级事务和子事务的中止。通过跟踪子事务的偏移量来实现这一点，
 * 然后使用这些偏移量来截断包含序列化更改的文件。
 *
 * 默认情况下，这些文件被放在临时文件目录中，文件名包含顶级事务的 XID 和
 * 订阅的 OID。这是必要的，以便处理具有相同 XID 的不同工作者不相互干扰。
 *
 * 我们使用 BufFiles 而不是正常的临时文件，因为 (a) BufFile 基础设施支持超过操作系统文件大小限制的临时文件，
 * (b) 提供了一种在发生错误时自动清理的方式，以及 (c) 提供了一种在本地事务中幸存这些文件的方法，并允许在流开启和关闭时打开和关闭它们。我们决定使用 FileSet
 * 基础设施，因为否则它会在文件关闭时删除文件，如果我们决定在开始/停止流时保持流文件打开，
 * 则会消耗大量内存（每个 BufFile 超过 8K，并且可能存在多个这样的 BufFiles，因为订阅者可能接收
 * 适用于不同事务的多个开始/停止流，随后才获得提交）。此外，如果我们不使用 FileSet，我们还需要发明
 * 通过 BufFile APIs 传递文件名的新方法，以便我们可以在多个流开启调用之间打开所需的文件
 * 用于同一事务。
 *
 * TWO_PHASE TRANSACTIONS
 * ----------------------
 * 两阶段事务在准备阶段重放，然后在提交准备和回滚准备时分别提交或回滚。可能会有一个准备事务在
 * apply 工作进程忙于执行初始复制时到达。在这种情况下，apply 工作进程将在 tablesync 仍然忙碌时跳过所有准备操作 [例如，插入] （请参见 should_apply_changes_for_rel 的条件）。
 * tablesync 工作进程可能不会得到这样的准备事务，因为假设它是在初始一致点之前到达，但可能得到了某些后来的提交。
 * 现在，tablesync 工作进程将退出，而不对被 apply 工作进程跳过的准备事务执行任何操作，因为它的同步位置将早于
 * apply 工作进程的当前的位置。这将导致一个“空准备”，因为稍后当 apply 工作进程执行提交准备时，里面没有任何内容（插入之前被跳过）。
 *
 * 为了避免这种情况以及类似的准备混淆，只有在初始同步完成后，订阅的两阶段提交才会启用。两阶段选项
 * 被实现为三态，值为 DISABLED、PENDING 和 ENABLED。
 *
 * 即使用户指定他们希望订阅的两阶段 = 开启，内部也会以 PENDING 的三态开始，只有在所有 tablesync 初始化完成后才变为 ENABLED - 也就是说，当所有
 * tablesync 工作进程都达到 READY 状态时。换句话说，PENDING 这个值只是订阅启动的临时状态。
 *
 * 在两阶段被正确可用（ENABLED）之前，订阅的行为就像两阶段 = 关闭。当 apply 工作进程检测到所有
 * tablesync 已经变为 READY（当三态为 PENDING）时，它将重新启动 apply 工作进程。这个过程发生在
 * process_syncing_tables_for_apply。
 *
 * 当（重新启动的）apply 工作进程发现所有 tablesyncs 都为一个两阶段三态 PENDING 时，它开始通过
 * 两阶段选项流式发送消息，这反过来使得在发布者处启用两阶段提交的解码。然后，它将三态值从 PENDING 更新为 ENABLED。
 * 现在，有可能在我们还没有启用两阶段的期间，发布者（复制服务器）会跳过一些准备，但我们确保这些准备与提交准备一起发送，见
 * ReorderBufferFinishPrepared。
 *
 * 如果订阅没有表，则两阶段三态 PENDING 不变。这使用户仍然可以执行 ALTER SUBSCRIPTION REFRESH
 * PUBLICATION，否则可能会被禁止（见下文）。
 *
 * 如果用户需要注意三态值，他们可以从 pg_subscription 目录中获取它（见列 subtwophasestate）。
 *
 * 我们不允许切换订阅的两阶段选项，因为这可能导致不一致的副本。考虑一下，最初它是开启的，
 * 我们已经接收到了一些准备，然后我们将其关闭，现在在提交时服务器将发送整个事务数据和提交。
 * 通过进一步的分析，我们可以允许将此选项从关闭更改为开启，但不确定这是否单独有用。
 *
 * 最后，为了避免前面段落中提到的问题，由于任何后续（未 READY）tablesync（需要将两阶段选项从“开启”切换到“关闭”，然后再切换回“开启”），
 * 对 ALTER SUBSCRIPTION REFRESH PUBLICATION 有一个限制。仅当 copy_data = false 时，此命令在两阶段三态为 ENABLED 时是不被允许的。
 *
 * 我们可能会针对真正的情况下获得同一 GID 的多个准备，那些情况下，我们已经为同一服务器上的
 * 出版物定义了多个订阅，而准备事务在这些订阅所接收的表上具有操作。对于这种情况，
 * 如果我们使用发布者发送的 GID，其中一个准备将会成功，而其他的将失败，服务器会再次发送它们。
 * 现在，如果用户为订阅者上的所有订阅设置了 synchronous_standby_names，这可能会导致死锁。为了避免这样的死锁，
 * 我们为每个准备事务在订阅者上生成一个唯一的 GID（由订阅的 oid 和准备事务的 xid 组成）。
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <sys/stat.h>
#include <unistd.h>

#include "access/table.h"
#include "access/tableam.h"
#include "access/twophase.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_tablespace.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/execPartition.h"
#include "executor/nodeModifyTable.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "postmaster/postmaster.h"
#include "postmaster/walwriter.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/logicalproto.h"
#include "replication/logicalrelation.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/dynahash.h"
#include "utils/datum.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/rls.h"
#include "utils/syscache.h"
#include "utils/timeout.h"

#define NAPTIME_PER_CYCLE 1000	/* 最大睡眠时间（循环之间）为1秒 */

typedef struct FlushPosition
{
	dlist_node	node;
	XLogRecPtr	local_end;
	XLogRecPtr	remote_end;
} FlushPosition;

static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);

typedef struct ApplyExecutionData
{
	EState	   *estate;			/* 执行器状态，用于跟踪资源 */

	LogicalRepRelMapEntry *targetRel;	/* 复制目标关系 */
	ResultRelInfo *targetRelInfo;	/* 相同的结果关系信息 */

	/* 当目标关系为分区时使用这些字段： */
	ModifyTableState *mtstate;	/* 虚拟的修改表状态 */
	PartitionTupleRouting *proute;	/* 分区路由信息 */
} ApplyExecutionData;

/* 用于保存和恢复应用错误上下文信息的结构 */
typedef struct ApplyErrorCallbackArg
{
	LogicalRepMsgType command;	/* 如果无效则为0 */
	LogicalRepRelMapEntry *rel;

	/* 远程节点信息 */
	int			remote_attnum;	/* 如果无效则为-1 */
	TransactionId remote_xid;
	XLogRecPtr	finish_lsn;
	char	   *origin_name;
} ApplyErrorCallbackArg;

static ApplyErrorCallbackArg apply_error_callback_arg =
{
	.command = 0,
	.rel = NULL,
	.remote_attnum = -1,
	.remote_xid = InvalidTransactionId,
	.finish_lsn = InvalidXLogRecPtr,
	.origin_name = NULL,
};

#ifdef FDDM
/* 在应用期间，持有远程事务的xid */
//用来记录远端事务id
TransactionId replication_origin_xid = InvalidTransactionId;


static bool is_multi_rep_sub = false;

sub_is_normal_hook_type   sub_judge_normal_hook = NULL;
handle_conflict_hook_type conflict_handle_hook = NULL;
table_missing_conflict_hook_type  table_missing_handle_hook = NULL;
find_partition_info_hook_type     fdd_partition_handle_hook = NULL;
column_missing_conflict_hook_type fdd_column_missing_handle_hook = NULL;

get_forward_subs_hook_type fdd_get_forward_subs_hook = NULL;


//update的时候，对端是否带了旧数据
bool rep_upadte_has_old = false;

//冲突处理需要跳过整个事务
bool fdd_skip_transaction =  false;
#endif

static MemoryContext ApplyMessageContext = NULL;
MemoryContext ApplyContext = NULL;

/* 用于流式事务的每个流上下文 */
static MemoryContext LogicalStreamingContext = NULL;

WalReceiverConn *LogRepWorkerWalRcvConn = NULL;

Subscription *MySubscription = NULL;
static bool MySubscriptionValid = false;

bool		in_remote_transaction = false;
#ifdef FDDM
XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
#else
static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
#endif
/* 仅在处理流式事务时有效的字段 */
static bool in_streamed_transaction = false;

static TransactionId stream_xid = InvalidTransactionId;

/*
 * 如果远程事务的完成LSN与subskiplsn匹配，
 * 则我们为订阅启用跳过所有数据修改更改（INSERT、UPDATE等）。
 * 一旦我们开始跳过更改，即使在那期间pg_subscription被更新，
 * MySubscription->skiplsn被改变或重置，我们也不会停止跳过，
 * 直到我们跳过所有该事务的更改。此外，在流式事务的情况下，
 * 我们不会跳过接收和缓存更改，因为我们决定在开始应用更改时
 * 是否跳过应用更改。成功跳过事务或应用非空事务后，清除subskiplsn。
 * 后者防止错误指定的subskiplsn被留下。
 */
static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
#define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))

/* 当前流式文件的BufFile句柄 */
static BufFile *stream_fd = NULL;

typedef struct SubXactInfo
{
	TransactionId xid;			/* 子事务的XID */
	int			fileno;			/* Buffile中的文件编号 */
	off_t		offset;			/* 文件中的偏移量 */
} SubXactInfo;

/* 当前流式事务的子事务数据 */
typedef struct ApplySubXactData
{
	uint32		nsubxacts;		/* 子事务数量 */
	uint32		nsubxacts_max;	/* 子事务的当前容量 */
	TransactionId subxact_last; /* 最后一个子事务的xid */
	SubXactInfo *subxacts;		/* 在更改文件中的子事务偏移量 */
} ApplySubXactData;

static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL};

static inline void fc_subxact_filename(char *fc_path, Oid fc_subid, TransactionId fc_xid);
static inline void fc_changes_filename(char *fc_path, Oid fc_subid, TransactionId fc_xid);

/*
 * 关于给定顶级事务的子事务的信息。
 */
static void fc_subxact_info_write(Oid fc_subid, TransactionId fc_xid);
static void fc_subxact_info_read(Oid fc_subid, TransactionId fc_xid);
static void fc_subxact_info_add(TransactionId fc_xid);
static inline void fc_cleanup_subxact_info(void);

/*
 * 为顶级事务序列化和反序列化更改。
 */
static void fc_stream_cleanup_files(Oid fc_subid, TransactionId fc_xid);
static void fc_stream_open_file(Oid fc_subid, TransactionId fc_xid, bool fc_first);
static void fc_stream_write_change(char fc_action, StringInfo fc_s);
static void fc_stream_close_file(void);

static void fc_send_feedback(XLogRecPtr fc_recvpos, bool fc_force, bool fc_requestReply);

static void fc_store_flush_position(XLogRecPtr fc_remote_lsn);

static void fc_maybe_reread_subscription(void);

static void fc_DisableSubscriptionAndExit(void);

/* 由于stream_commit需要的原型 */
static void fc_apply_dispatch(StringInfo fc_s);

static void fc_apply_handle_commit_internal(LogicalRepCommitData *fc_commit_data);
static void fc_apply_handle_insert_internal(ApplyExecutionData *fc_edata,
										 ResultRelInfo *fc_relinfo,
										 TupleTableSlot *fc_remoteslot);
static void fc_apply_handle_update_internal(ApplyExecutionData *fc_edata,
										 ResultRelInfo *fc_relinfo,
										 TupleTableSlot *fc_remoteslot,
										 LogicalRepTupleData *fc_newtup);
static void fc_apply_handle_delete_internal(ApplyExecutionData *fc_edata,
										 ResultRelInfo *fc_relinfo,
										 TupleTableSlot *fc_remoteslot);
static bool fc_FindReplTupleInLocalRel(EState *fc_estate, Relation fc_localrel,
									LogicalRepRelation *fc_remoterel,
									TupleTableSlot *fc_remoteslot,
									TupleTableSlot **fc_localslot);
static void fc_apply_handle_tuple_routing(ApplyExecutionData *fc_edata,
									   TupleTableSlot *fc_remoteslot,
									   LogicalRepTupleData *fc_newtup,
									   CmdType fc_operation);

/* 计算两阶段事务的GID */
static void fc_TwoPhaseTransactionGid(Oid fc_subid, TransactionId fc_xid, char *fc_gid, int fc_szgid);

/* 应用所有缓存消息的通用流式函数 */
static void fc_apply_spooled_messages(TransactionId fc_xid, XLogRecPtr fc_lsn);

/* 跳过更改的函数 */
static void fc_maybe_start_skipping_changes(XLogRecPtr fc_finish_lsn);
static void fc_stop_skipping_changes(void);
static void fc_clear_subscription_skip_lsn(XLogRecPtr fc_finish_lsn);

/* 应用错误回调的函数 */
static void fc_apply_error_callback(void *fc_arg);
static inline void fc_set_apply_error_context_xact(TransactionId fc_xid, XLogRecPtr fc_lsn);
static inline void fc_reset_apply_error_context_info(void);

/*
 * 此工作者是否应为给定关系应用更改。
 *
 * 这主要是在初始关系数据同步时需要，因为这在
 * 另一个并行运行的工作进程中进行，我们需要某种方式跳过
 * 在表的同步过程中发送到主应用工作者的更改。
 *
 * 请注意，我们需要对SYNCDONE状态进行小于或等于的比较，
 * 因为它可能持有初始插槽一致性点WAL记录+1（即下一个记录的开始）
 * 的结束位置，而下一个记录可能是我们现在处理的事务的提交
 * （这就是我们在apply_handle_begin中设置remote_final_lsn的原因）。
 */
static bool fc_should_apply_changes_for_rel(LogicalRepRelMapEntry *fc_rel)
{
	if (am_tablesync_worker())
		return MyLogicalRepWorker->relid == fc_rel->localreloid;
	else
		return (fc_rel->state == SUBREL_STATE_READY ||
				(fc_rel->state == SUBREL_STATE_SYNCDONE &&
				 fc_rel->statelsn <= remote_final_lsn));
}

/*
 * 开始一个复制事务的一个步骤（一个 INSERT、UPDATE 等）。
 *
 * 如果这是第一步，则开始一个事务（否则我们继续使用
 * 现有事务）。
 * 同时提供一个全局快照，并确保我们在 ApplyMessageContext 中运行。
 */
static void fc_begin_replication_step(void)
{
	SetCurrentStatementStartTimestamp();

	if (!IsTransactionState())
	{
		StartTransactionCommand();
		fc_maybe_reread_subscription();
	}

	PushActiveSnapshot(GetTransactionSnapshot());

	MemoryContextSwitchTo(ApplyMessageContext);
}

/*
 * 完成一个复制事务的一个步骤。
 * 调用 begin_replication_step() 的调用者也必须调用此函数。
 *
 * 这里我们不关闭事务，但我们应该递增
 * 命令计数器以使该步骤的效果可见。
 */
static void fc_end_replication_step(void)
{
	PopActiveSnapshot();

	CommandCounterIncrement();
}

/*
 * 处理流式事务。
 *
 * 如果处于流式模式（接收一块流式事务），我们
 * 只需将其重定向到适当的顶层事务的文件即可。
 *
 * 对于流式事务返回 true，否则返回 false（常规模式）。
 */
static bool fc_handle_streamed_transaction(LogicalRepMsgType fc_action, StringInfo fc_s)
{
	TransactionId fc_xid;

	/* 不在流式模式中 */
	if (!in_streamed_transaction)
		return false;

	Assert(stream_fd != NULL);
	Assert(TransactionIdIsValid(stream_xid));

	/*
	 * 我们应该已经收到了子事务的 XID 作为消息的第一部分，
	 * 因此提取它。
	 */
	fc_xid = pq_getmsgint(fc_s, 4);

	if (!TransactionIdIsValid(fc_xid))
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("invalid transaction ID in streamed replication transaction")));

	/* 将新的子事务添加到数组（除非已经存在）。 */
	fc_subxact_info_add(fc_xid);

	/* 将更改写入当前文件 */
	fc_stream_write_change(fc_action, fc_s);

	return true;
}

/*
 * 为指定关系的约束表达式、
 * 索引和触发器评估准备执行者状态。
 *
 * 注意，调用者必须打开并关闭任何要更新的索引。
 */
static ApplyExecutionData *
fc_create_edata_for_relation(LogicalRepRelMapEntry *fc_rel)
{
	ApplyExecutionData *fc_edata;
	EState	   *fc_estate;
	RangeTblEntry *fc_rte;
	ResultRelInfo *fc_resultRelInfo;

	fc_edata = (ApplyExecutionData *) palloc0(sizeof(ApplyExecutionData));
	fc_edata->targetRel = fc_rel;

	fc_edata->estate = fc_estate = CreateExecutorState();

	fc_rte = makeNode(RangeTblEntry);
	fc_rte->rtekind = RTE_RELATION;
	fc_rte->relid = RelationGetRelid(fc_rel->localrel);
	fc_rte->relkind = fc_rel->localrel->rd_rel->relkind;
	fc_rte->rellockmode = AccessShareLock;
	ExecInitRangeTable(fc_estate, list_make1(fc_rte));

	fc_edata->targetRelInfo = fc_resultRelInfo = makeNode(ResultRelInfo);

	/*
	 * 使用 logicalrep_rel_open() 打开的关系，而不是再次打开它。
	 */
	InitResultRelInfo(fc_resultRelInfo, fc_rel->localrel, 1, NULL, 0);

	/*
	 * 我们将 ResultRelInfo 放入 es_opened_result_relations 列表中，
	 * 即使我们没有填充 es_result_relations 数组。这有点
	 * 不太对，但足以让 ExecGetTriggerResultRel() 找到它们。
	 *
	 * 这里也没有调用 ExecOpenIndices()，每个执行路径负责进行
	 * 应用操作。
	 */
	fc_estate->es_opened_result_relations =
		lappend(fc_estate->es_opened_result_relations, fc_resultRelInfo);

	fc_estate->es_output_cid = GetCurrentCommandId(true);

	/* 准备捕获 AFTER 触发器。 */
	AfterTriggerBeginQuery();

	/* edata 的其他字段暂时保持为 NULL */

	return fc_edata;
}

/*
 * 完成与由 create_edata_for_relation() 创建的执行者状态相关的任何操作。
 */
static void fc_finish_edata(ApplyExecutionData *fc_edata)
{
	EState	   *fc_estate = fc_edata->estate;

	/* 处理任何排队的 AFTER 触发器。 */
	AfterTriggerEndQuery(fc_estate);

	/* 如果进行了 tuple 路由，则关闭它。 */
	if (fc_edata->proute)
		ExecCleanupTupleRouting(fc_edata->mtstate, fc_edata->proute);

	/*
	 * 清理。看起来我们应该在这里调用 ExecCloseResultRelations()
	 * ，但我们故意不这样做。它会关闭我们在上面
	 * 添加到 es_opened_result_relations 的关系，这是错误的，因为我们没有相应的引用计数。
	 * 我们依赖 ExecCleanupTupleRouting() 来关闭
	 * 在执行期间打开的其他关系。
	 */
	ExecResetTupleTable(fc_estate->es_tupleTable, false);
	FreeExecutorState(fc_estate);
	pfree(fc_edata);
}

/*
 * 对于我们无法映射到远程关系列的列执行默认值。
 *
 * 这使我们能够支持下游表具有比上游更多的列。
 */
static void fc_slot_fill_defaults(LogicalRepRelMapEntry *fc_rel, EState *fc_estate,
				   TupleTableSlot *fc_slot)
{
	TupleDesc	fc_desc = RelationGetDescr(fc_rel->localrel);
	int			fc_num_phys_attrs = fc_desc->natts;
	int			fc_i;
	int			fc_attnum,
				fc_num_defaults = 0;
	int		   *fc_defmap;
	ExprState **fc_defexprs;
	ExprContext *fc_econtext;

	fc_econtext = GetPerTupleExprContext(fc_estate);

	/* 我们通过复制获取了所有数据，不需要评估任何内容。 */
	if (fc_num_phys_attrs == fc_rel->remoterel.natts)
		return;

	fc_defmap = (int *) palloc(fc_num_phys_attrs * sizeof(int));
	fc_defexprs = (ExprState **) palloc(fc_num_phys_attrs * sizeof(ExprState *));

	Assert(fc_rel->attrmap->maplen == fc_num_phys_attrs);
	for (fc_attnum = 0; fc_attnum < fc_num_phys_attrs; fc_attnum++)
	{
		Expr	   *fc_defexpr;

		if (TupleDescAttr(fc_desc, fc_attnum)->attisdropped || TupleDescAttr(fc_desc, fc_attnum)->attgenerated)
			continue;

		if (fc_rel->attrmap->attnums[fc_attnum] >= 0)
			continue;

		fc_defexpr = (Expr *) build_column_default(fc_rel->localrel, fc_attnum + 1);

		if (fc_defexpr != NULL)
		{
			/* 通过计划运行表达式 */
			fc_defexpr = expression_planner(fc_defexpr);

			/* 在复制上下文中初始化可执行表达式 */
			fc_defexprs[fc_num_defaults] = ExecInitExpr(fc_defexpr, NULL);
			fc_defmap[fc_num_defaults] = fc_attnum;
			fc_num_defaults++;
		}
	}

	for (fc_i = 0; fc_i < fc_num_defaults; fc_i++)
		fc_slot->tts_values[fc_defmap[fc_i]] =
			ExecEvalExpr(fc_defexprs[fc_i], fc_econtext, &fc_slot->tts_isnull[fc_defmap[fc_i]]);
}

/*
 * 将元组数据存储到插槽中。
 *
 * 输入数据可以是文本或二进制格式。
 */
static void fc_slot_store_data(TupleTableSlot *fc_slot, LogicalRepRelMapEntry *fc_rel,
				LogicalRepTupleData *fc_tupleData)
{
	int			fc_natts = fc_slot->tts_tupleDescriptor->natts;
	int			fc_i;

	ExecClearTuple(fc_slot);

	/* 对每个未丢弃且非空的属性调用 "in" 函数 */
	Assert(fc_natts == fc_rel->attrmap->maplen);
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_slot->tts_tupleDescriptor, fc_i);
		int			fc_remoteattnum = fc_rel->attrmap->attnums[fc_i];

		if (!fc_att->attisdropped && fc_remoteattnum >= 0)
		{
			StringInfo	fc_colvalue = &fc_tupleData->colvalues[fc_remoteattnum];

			Assert(fc_remoteattnum < fc_tupleData->ncols);

			/* 为错误回调设置 attnum */
			apply_error_callback_arg.remote_attnum = fc_remoteattnum;

			if (fc_tupleData->colstatus[fc_remoteattnum] == LOGICALREP_COLUMN_TEXT)
			{
				Oid			fc_typinput;
				Oid			fc_typioparam;

				getTypeInputInfo(fc_att->atttypid, &fc_typinput, &fc_typioparam);
				fc_slot->tts_values[fc_i] =
					OidInputFunctionCall(fc_typinput, fc_colvalue->data,
										 fc_typioparam, fc_att->atttypmod);
				fc_slot->tts_isnull[fc_i] = false;
			}
			else if (fc_tupleData->colstatus[fc_remoteattnum] == LOGICALREP_COLUMN_BINARY)
			{
				Oid			fc_typreceive;
				Oid			fc_typioparam;

				/*
				 * 在某些代码路径中，我们可能会被要求重新解析相同的
				 * 元组数据。重置 StringInfo 的游标，以便可以正常工作。
				 */
				fc_colvalue->cursor = 0;

				getTypeBinaryInputInfo(fc_att->atttypid, &fc_typreceive, &fc_typioparam);
				fc_slot->tts_values[fc_i] =
					OidReceiveFunctionCall(fc_typreceive, fc_colvalue,
										   fc_typioparam, fc_att->atttypmod);

				/* 如果没有读取完整个缓冲区则会出问题 */
				if (fc_colvalue->cursor != fc_colvalue->len)
					ereport(ERROR,
							(errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
							 errmsg("incorrect binary data format in logical replication column %d",
									fc_remoteattnum + 1)));
				fc_slot->tts_isnull[fc_i] = false;
			}
			else
			{
				/*
				 * 来自远程的 NULL 值。 （我们不希望在这里看到
				 * LOGICALREP_COLUMN_UNCHANGED，但如果看到，将其视为
				 * NULL。）
				 */
				fc_slot->tts_values[fc_i] = (Datum) 0;
				fc_slot->tts_isnull[fc_i] = true;
			}

			/* 重置错误回调的 attnum */
			apply_error_callback_arg.remote_attnum = -1;
		}
		else
		{
			/*
			 * 我们将 NULL 分配给被丢弃的属性和缺失的值
			 * （缺失的值应在后续使用 slot_fill_defaults 填充）。
			 */
			fc_slot->tts_values[fc_i] = (Datum) 0;
			fc_slot->tts_isnull[fc_i] = true;
		}
	}

	ExecStoreVirtualTuple(fc_slot);
}

/*
 * 用 LogicalRepTupleData 结构中的数据替换更新的列。
 * 这在某种程度上类似于 heap_modify_tuple，但也会对用户数据调用类型
 * 输入函数。
 *
 * “slot” 用 srcslot 中的元组副本填充，替换
 * “tupleData”中提供的列，其他列保持不变。
 *
 * 警告：在“slot”中未替换的按引用传递列将指向
 * “srcslot”的存储。这在当前使用中是可以的，但将来我们可能需要
 * 在结束时物化“slot”，使其独立于“srcslot”。
 */
static void fc_slot_modify_data(TupleTableSlot *fc_slot, TupleTableSlot *fc_srcslot,
				 LogicalRepRelMapEntry *fc_rel,
				 LogicalRepTupleData *fc_tupleData)
{
	int			fc_natts = fc_slot->tts_tupleDescriptor->natts;
	int			fc_i;

	/* 我们将用一个虚拟元组填充“slot”，因此必须从...开始 */
	ExecClearTuple(fc_slot);

	/*
	 * 从 srcslot 中复制所有列数据，以便我们将拥有有效的值
	 * 用于未替换的列。
	 */
	Assert(fc_natts == fc_srcslot->tts_tupleDescriptor->natts);
	slot_getallattrs(fc_srcslot);
	memcpy(fc_slot->tts_values, fc_srcslot->tts_values, fc_natts * sizeof(Datum));
	memcpy(fc_slot->tts_isnull, fc_srcslot->tts_isnull, fc_natts * sizeof(bool));

	/* 调用每个替换属性的“in”函数 */
	Assert(fc_natts == fc_rel->attrmap->maplen);
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_slot->tts_tupleDescriptor, fc_i);
		int			fc_remoteattnum = fc_rel->attrmap->attnums[fc_i];

		if (fc_remoteattnum < 0)
			continue;

		Assert(fc_remoteattnum < fc_tupleData->ncols);

		if (fc_tupleData->colstatus[fc_remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
		{
			StringInfo	fc_colvalue = &fc_tupleData->colvalues[fc_remoteattnum];

			/* 为错误回调设置 attnum */
			apply_error_callback_arg.remote_attnum = fc_remoteattnum;

			if (fc_tupleData->colstatus[fc_remoteattnum] == LOGICALREP_COLUMN_TEXT)
			{
				Oid			fc_typinput;
				Oid			fc_typioparam;

				getTypeInputInfo(fc_att->atttypid, &fc_typinput, &fc_typioparam);
				fc_slot->tts_values[fc_i] =
					OidInputFunctionCall(fc_typinput, fc_colvalue->data,
										 fc_typioparam, fc_att->atttypmod);
				fc_slot->tts_isnull[fc_i] = false;
			}
			else if (fc_tupleData->colstatus[fc_remoteattnum] == LOGICALREP_COLUMN_BINARY)
			{
				Oid			fc_typreceive;
				Oid			fc_typioparam;

				/*
				 * 在某些代码路径中，我们可能会被要求重新解析相同的
				 * 元组数据。重置 StringInfo 的游标，以便可以正常工作。
				 */
				fc_colvalue->cursor = 0;

				getTypeBinaryInputInfo(fc_att->atttypid, &fc_typreceive, &fc_typioparam);
				fc_slot->tts_values[fc_i] =
					OidReceiveFunctionCall(fc_typreceive, fc_colvalue,
										   fc_typioparam, fc_att->atttypmod);

				/* 如果没有读取完整个缓冲区则会出问题 */
				if (fc_colvalue->cursor != fc_colvalue->len)
					ereport(ERROR,
							(errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
							 errmsg("incorrect binary data format in logical replication column %d",
									fc_remoteattnum + 1)));
				fc_slot->tts_isnull[fc_i] = false;
			}
			else
			{
				/* 必须是 LOGICALREP_COLUMN_NULL */
				fc_slot->tts_values[fc_i] = (Datum) 0;
				fc_slot->tts_isnull[fc_i] = true;
			}

			/* 重置错误回调的 attnum */
			apply_error_callback_arg.remote_attnum = -1;
		}
	}

	/* 最后，声明“slot”包含一个有效的虚拟元组 */
	ExecStoreVirtualTuple(fc_slot);
}

/*
 * 处理 BEGIN 消息。
 */
static void fc_apply_handle_begin(StringInfo fc_s)
{
	LogicalRepBeginData fc_begin_data;

	logicalrep_read_begin(fc_s, &fc_begin_data);
	fc_set_apply_error_context_xact(fc_begin_data.xid, fc_begin_data.final_lsn);

	remote_final_lsn = fc_begin_data.final_lsn;

	fc_maybe_start_skipping_changes(fc_begin_data.final_lsn);

	in_remote_transaction = true;
#ifdef FDDM
	//记录远端事务id
	replication_origin_xid = fc_begin_data.xid;
	replorigin_session_origin_timestamp = fc_begin_data.committime;
	fdd_skip_transaction = false;
#endif

	pgstat_report_activity(STATE_RUNNING, NULL);
}

/*
 * 处理 COMMIT 消息。
 *
 * TODO，支持跟踪多个来源
 */
static void fc_apply_handle_commit(StringInfo fc_s)
{
	LogicalRepCommitData fc_commit_data;

#ifdef FDDM
	if (fdd_skip_transaction)
	{
		ereport(WARNING, (errmsg("current transaction has been setted to skip by conflict handler.")));
		AbortCurrentTransaction();
	}
#endif

	logicalrep_read_commit(fc_s, &fc_commit_data);

	if (fc_commit_data.commit_lsn != remote_final_lsn)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("incorrect commit LSN %X/%X in commit message (expected %X/%X)",
								 LSN_FORMAT_ARGS(fc_commit_data.commit_lsn),
								 LSN_FORMAT_ARGS(remote_final_lsn))));

	fc_apply_handle_commit_internal(&fc_commit_data);

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_commit_data.end_lsn);
#ifdef FDDM
	//清理远端事务id
	replication_origin_xid = InvalidTransactionId;
#endif

	pgstat_report_activity(STATE_IDLE, NULL);
	fc_reset_apply_error_context_info();
}

/*
 * 处理 BEGIN PREPARE 消息。
 */
static void fc_apply_handle_begin_prepare(StringInfo fc_s)
{
	LogicalRepPreparedTxnData fc_begin_data;

	/* Tablesync 永远不应接收 prepare。 */
	if (am_tablesync_worker())
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("tablesync worker received a BEGIN PREPARE message")));

	logicalrep_read_begin_prepare(fc_s, &fc_begin_data);
	fc_set_apply_error_context_xact(fc_begin_data.xid, fc_begin_data.prepare_lsn);

	remote_final_lsn = fc_begin_data.prepare_lsn;

	fc_maybe_start_skipping_changes(fc_begin_data.prepare_lsn);

	in_remote_transaction = true;

	pgstat_report_activity(STATE_RUNNING, NULL);
}

/*
 * 准备 GID 的通用函数。
 */
static void fc_apply_handle_prepare_internal(LogicalRepPreparedTxnData *fc_prepare_data)
{
	char		fc_gid[GIDSIZE];

	/*
	 * 为两阶段事务计算唯一 GID。我们不使用
	 * 服务器发送的已准备事务的 GID，因为这可能会导致死锁，当
	 * 我们有多个订阅从同一节点指向同一节点上的出版物时。请参见 worker.c 顶部的评论。
	 */
	fc_TwoPhaseTransactionGid(MySubscription->oid, fc_prepare_data->xid,
						   fc_gid, sizeof(fc_gid));

	/*
	 * BeginTransactionBlock是平衡下面的 PrepareTransactionBlock 中调用的
	 * EndTransactionBlock 的必要步骤。
	 */
	BeginTransactionBlock();
	CommitTransactionCommand(); /* 完成前面的 Begin 命令。 */

	/*
	 * 更新原始状态，以便在崩溃的情况下我们可以从正确的位置
	 * 重新启动流。
	 */
	replorigin_session_origin_lsn = fc_prepare_data->end_lsn;
	replorigin_session_origin_timestamp = fc_prepare_data->prepare_time;

	PrepareTransactionBlock(fc_gid);
}

/*
 * 处理 PREPARE 消息。
 */
static void fc_apply_handle_prepare(StringInfo fc_s)
{
	LogicalRepPreparedTxnData fc_prepare_data;

	logicalrep_read_prepare(fc_s, &fc_prepare_data);

	if (fc_prepare_data.prepare_lsn != remote_final_lsn)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("incorrect prepare LSN %X/%X in prepare message (expected %X/%X)",
								 LSN_FORMAT_ARGS(fc_prepare_data.prepare_lsn),
								 LSN_FORMAT_ARGS(remote_final_lsn))));

	/*
	 * 与提交不同，在这里，即使该事务中没有发生任何更改或所有更改被跳过，我们总是会准备事务。这样做是因为在提交准备时，我们无法知道是否由于这些原因跳过了准备事务。
	 *
	 * XXX，我们可以优化，在提交准备时，首先检查我们是否已经准备了事务，但这似乎不值得，因为这种情况不应该是常见的。
	 */
	fc_begin_replication_step();

	fc_apply_handle_prepare_internal(&fc_prepare_data);

	fc_end_replication_step();
	CommitTransactionCommand();
	pgstat_report_stat(false);

	fc_store_flush_position(fc_prepare_data.end_lsn);

	in_remote_transaction = false;

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_prepare_data.end_lsn);

	/*
	 * 由于我们已经准备了事务，在服务器在清除subskiplsn之前崩溃的情况下，subskiplsn将被保留，但事务不会被重新发送。但这没关系，因为这是一个罕见的情况，并且在完成下一个事务时subskiplsn将被清除。
	 */
	fc_stop_skipping_changes();
	fc_clear_subscription_skip_lsn(fc_prepare_data.prepare_lsn);

	pgstat_report_activity(STATE_IDLE, NULL);
	fc_reset_apply_error_context_info();
}

/*
 * 处理先前准备的事务的COMMIT PREPARED。
 */
static void fc_apply_handle_commit_prepared(StringInfo fc_s)
{
	LogicalRepCommitPreparedTxnData fc_prepare_data;
	char		fc_gid[GIDSIZE];

	logicalrep_read_commit_prepared(fc_s, &fc_prepare_data);
	fc_set_apply_error_context_xact(fc_prepare_data.xid, fc_prepare_data.commit_lsn);

	/* 计算两阶段事务的GID。 */
	fc_TwoPhaseTransactionGid(MySubscription->oid, fc_prepare_data.xid,
						   fc_gid, sizeof(fc_gid));

	/* 当调用COMMIT PREPARED时，没有事务 */
	fc_begin_replication_step();

	/*
	 * 更新原始状态，以便在崩溃的情况下我们可以从正确的位置
	 * 重新启动流。
	 */
	replorigin_session_origin_lsn = fc_prepare_data.end_lsn;
	replorigin_session_origin_timestamp = fc_prepare_data.commit_time;

	FinishPreparedTransaction(fc_gid, true);
	fc_end_replication_step();
	CommitTransactionCommand();
	pgstat_report_stat(false);

	fc_store_flush_position(fc_prepare_data.end_lsn);
	in_remote_transaction = false;

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_prepare_data.end_lsn);

	fc_clear_subscription_skip_lsn(fc_prepare_data.end_lsn);

	pgstat_report_activity(STATE_IDLE, NULL);
	fc_reset_apply_error_context_info();
}

/*
 * 处理先前准备的事务的ROLLBACK PREPARED。
 */
static void fc_apply_handle_rollback_prepared(StringInfo fc_s)
{
	LogicalRepRollbackPreparedTxnData fc_rollback_data;
	char		fc_gid[GIDSIZE];

	logicalrep_read_rollback_prepared(fc_s, &fc_rollback_data);
	fc_set_apply_error_context_xact(fc_rollback_data.xid, fc_rollback_data.rollback_end_lsn);

	/* 计算两阶段事务的GID。 */
	fc_TwoPhaseTransactionGid(MySubscription->oid, fc_rollback_data.xid,
						   fc_gid, sizeof(fc_gid));

	/*
	 * 有可能我们没有收到准备，因为它发生在walsender到达一致点之前，或者在那时两阶段尚未启用，因此在这种情况下，我们需要跳过回滚准备。
	 */
	if (LookupGXact(fc_gid, fc_rollback_data.prepare_end_lsn,
					fc_rollback_data.prepare_time))
	{
		/*
		 * 更新原始状态，以便在崩溃的情况下可以从正确的位置重新启动流媒体。
		 */
		replorigin_session_origin_lsn = fc_rollback_data.rollback_end_lsn;
		replorigin_session_origin_timestamp = fc_rollback_data.rollback_time;

		/* 当调用ABORT/ROLLBACK PREPARED时，没有事务 */
		fc_begin_replication_step();
		FinishPreparedTransaction(fc_gid, false);
		fc_end_replication_step();
		CommitTransactionCommand();

		fc_clear_subscription_skip_lsn(fc_rollback_data.rollback_end_lsn);
	}

	pgstat_report_stat(false);

	fc_store_flush_position(fc_rollback_data.rollback_end_lsn);
	in_remote_transaction = false;

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_rollback_data.rollback_end_lsn);

	pgstat_report_activity(STATE_IDLE, NULL);
	fc_reset_apply_error_context_info();
}

/*
 * 处理STREAM PREPARE。
 *
 * 逻辑分为两个部分：
 * 1. 重放所有缓冲的操作
 * 2. 将事务标记为已准备
 */
static void fc_apply_handle_stream_prepare(StringInfo fc_s)
{
	LogicalRepPreparedTxnData fc_prepare_data;

	if (in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("STREAM PREPARE message without STREAM STOP")));

	/* Tablesync 永远不应接收 prepare。 */
	if (am_tablesync_worker())
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));

	logicalrep_read_stream_prepare(fc_s, &fc_prepare_data);
	fc_set_apply_error_context_xact(fc_prepare_data.xid, fc_prepare_data.prepare_lsn);

	elog(DEBUG1, "received prepare for streamed transaction %u", fc_prepare_data.xid);

	/* 重放所有缓冲的操作。 */
	fc_apply_spooled_messages(fc_prepare_data.xid, fc_prepare_data.prepare_lsn);

	/* 将事务标记为已准备。 */
	fc_apply_handle_prepare_internal(&fc_prepare_data);

	CommitTransactionCommand();

	pgstat_report_stat(false);

	fc_store_flush_position(fc_prepare_data.end_lsn);

	in_remote_transaction = false;

	/* 解除序列化更改和子事务信息的文件链接。 */
	fc_stream_cleanup_files(MyLogicalRepWorker->subid, fc_prepare_data.xid);

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_prepare_data.end_lsn);

	/*
	 * 与准备案例类似，subskiplsn可能在服务器崩溃的情况下被保留，但这没关系。请参阅apply_handle_prepare()中的评论。
	 */
	fc_stop_skipping_changes();
	fc_clear_subscription_skip_lsn(fc_prepare_data.prepare_lsn);

	pgstat_report_activity(STATE_IDLE, NULL);

	fc_reset_apply_error_context_info();
}

/*
 * 处理ORIGIN消息。
 *
 * TODO，支持跟踪多个源
 */
static void fc_apply_handle_origin(StringInfo fc_s)
{
	/*
	 * ORIGIN消息只能在流媒体事务内或远程事务内，并且在任何实际写入之前出现。
	 */
	if (!in_streamed_transaction &&
		(!in_remote_transaction ||
		 (IsTransactionState() && !am_tablesync_worker())))
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("ORIGIN message sent out of order")));
}

/*
 * 处理STREAM START消息。
 */
static void fc_apply_handle_stream_start(StringInfo fc_s)
{
	bool		fc_first_segment;

	if (in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("duplicate STREAM START message")));

	/*
	 * 在流开始时启动事务，除非它是一个表同步工作者，否则该事务将在流停止时提交，
	 * 在这种情况下，它将在处理所有消息后提交。我们需要这个事务来处理
	 * buffile，用于序列化流数据和子事务信息。
	 */
	fc_begin_replication_step();

	/* 通知处理方法我们正在处理一个远程事务 */
	in_streamed_transaction = true;

	/* 提取顶级事务的XID */
	stream_xid = logicalrep_read_stream_start(fc_s, &fc_first_segment);

	if (!TransactionIdIsValid(stream_xid))
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("invalid transaction ID in streamed replication transaction")));

	fc_set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);

	/*
	 * 如果我们还没有初始化工作者的stream_fileset，就进行初始化。这将在工作者的整个持续时间内使用，
	 * 所以在一个持久上下文中创建它。我们在任何事务的第一个流消息上创建它，
	 * 然后将其用于这个和其他的流事务。
	 * 现在，我们可以在工作者开始时创建一个fileset，但那样我们就无法确保它会被使用。
	 */
	if (MyLogicalRepWorker->stream_fileset == NULL)
	{
		MemoryContext fc_oldctx;

		fc_oldctx = MemoryContextSwitchTo(ApplyContext);

		MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
		FileSetInit(MyLogicalRepWorker->stream_fileset);

		MemoryContextSwitchTo(fc_oldctx);
	}

	/* 为该事务打开临时文件 */
	fc_stream_open_file(MyLogicalRepWorker->subid, stream_xid, fc_first_segment);

	/* 如果这不是第一个片段，打开现有的子事务文件 */
	if (!fc_first_segment)
		fc_subxact_info_read(MyLogicalRepWorker->subid, stream_xid);

	pgstat_report_activity(STATE_RUNNING, NULL);

	fc_end_replication_step();
}

/*
 * 处理STREAM STOP消息。
 */
static void fc_apply_handle_stream_stop(StringInfo fc_s)
{
	if (!in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("STREAM STOP message without STREAM START")));

	/*
	 * 关闭序列化更改的文件，并序列化有关顶级事务的子事务信息。
	 */
	fc_subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
	fc_stream_close_file();

	/* 我们必须处于有效的事务状态 */
	Assert(IsTransactionState());

	/* 提交每个流的事务 */
	CommitTransactionCommand();

	in_streamed_transaction = false;

	/* 重置每个流的上下文 */
	MemoryContextReset(LogicalStreamingContext);

	pgstat_report_activity(STATE_IDLE, NULL);
	fc_reset_apply_error_context_info();
}

/*
 * 处理STREAM abort消息。
 */
static void fc_apply_handle_stream_abort(StringInfo fc_s)
{
	TransactionId fc_xid;
	TransactionId fc_subxid;

	if (in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("STREAM ABORT message without STREAM STOP")));

	logicalrep_read_stream_abort(fc_s, &fc_xid, &fc_subxid);

	/*
	 * 如果两个XID相同，实际上它是顶级事务的回滚，所以
	 * 只需删除序列化信息的文件。
	 */
	if (fc_xid == fc_subxid)
	{
		fc_set_apply_error_context_xact(fc_xid, InvalidXLogRecPtr);
		fc_stream_cleanup_files(MyLogicalRepWorker->subid, fc_xid);
	}
	else
	{
		/*
		 * 好吧，这是一个子事务。我们需要读取顶级事务的子事务文件，
		 * 确定为子事务跟踪的偏移量，
		 * 并截断包含更改的文件。我们还会移除偏移量更高（或者说XID更高）的子事务。
		 *
		 * 我们故意从尾部扫描数组，因为我们很可能
		 * 正在回滚最近的子事务的更改。
		 *
		 * 我们不能在这里使用二分查找，因为子事务的XID不一定
		 * 是按顺序到达的，考虑一下我们已经
		 * 为多个子事务释放保存点，然后
		 * 对其中一个早期的子事务执行回滚到保存点的情况。
		 */
		int64		fc_i;
		int64		fc_subidx;
		BufFile    *fc_fd;
		bool		fc_found = false;
		char		fc_path[MAXPGPATH];

		fc_set_apply_error_context_xact(fc_subxid, InvalidXLogRecPtr);

		fc_subidx = -1;
		fc_begin_replication_step();
		fc_subxact_info_read(MyLogicalRepWorker->subid, fc_xid);

		for (fc_i = subxact_data.nsubxacts; fc_i > 0; fc_i--)
		{
			if (subxact_data.subxacts[fc_i - 1].xid == fc_subxid)
			{
				fc_subidx = (fc_i - 1);
				fc_found = true;
				break;
			}
		}

		/*
		 * 如果这个子事务是空的，我们将无法在这里找到subxid，
		 * 所以只需清理子事务信息并返回。
		 */
		if (!fc_found)
		{
			/* 清理子事务信息 */
			fc_cleanup_subxact_info();
			fc_end_replication_step();
			CommitTransactionCommand();
			fc_reset_apply_error_context_info();
			return;
		}

		/* 打开更改文件 */
		fc_changes_filename(fc_path, MyLogicalRepWorker->subid, fc_xid);
		fc_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, fc_path,
								O_RDWR, false);

		/* 好的，在正确的偏移量截断文件 */
		BufFileTruncateFileSet(fc_fd, subxact_data.subxacts[fc_subidx].fileno,
							   subxact_data.subxacts[fc_subidx].offset);
		BufFileClose(fc_fd);

		/* 丢弃稍后添加的子事务 */
		subxact_data.nsubxacts = fc_subidx;

		/* 写入更新后的子事务列表 */
		fc_subxact_info_write(MyLogicalRepWorker->subid, fc_xid);

		fc_end_replication_step();
		CommitTransactionCommand();
	}

	fc_reset_apply_error_context_info();
}

/*
 * 通用临时文件处理。
 */
static void fc_apply_spooled_messages(TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	StringInfoData fc_s2;
	int			fc_nchanges;
	char		fc_path[MAXPGPATH];
	char	   *fc_buffer = NULL;
	MemoryContext fc_oldcxt;
	BufFile    *fc_fd;

	fc_maybe_start_skipping_changes(fc_lsn);

	/* 确保我们有一个打开的事务 */
	fc_begin_replication_step();

	/*
	 * 分配处理TopTransactionContext中所有消息所需的文件句柄和内存，
	 * 以避免在每个消息处理后重置它们。
	 */
	fc_oldcxt = MemoryContextSwitchTo(TopTransactionContext);

	/* 为提交/准备的事务打开临时文件 */
	fc_changes_filename(fc_path, MyLogicalRepWorker->subid, fc_xid);
	elog(DEBUG1, "replaying changes from file \"%s\"", fc_path);

	fc_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, fc_path, O_RDONLY,
							false);

	fc_buffer = palloc(BLCKSZ);
	initStringInfo(&fc_s2);

	MemoryContextSwitchTo(fc_oldcxt);

	remote_final_lsn = fc_lsn;

	/*
	 * 确保句柄的apply_dispatch方法知道我们在进行远程
	 * 事务。
	 */
	in_remote_transaction = true;
	pgstat_report_activity(STATE_RUNNING, NULL);

	fc_end_replication_step();

	/*
	 * 逐个读取条目，并按照apply_dispatch中的相同逻辑处理它们。
	 */
	fc_nchanges = 0;
	while (true)
	{
		size_t		fc_nbytes;
		int			fc_len;

		CHECK_FOR_INTERRUPTS();

		/* 读取磁盘记录的长度 */
		fc_nbytes = BufFileRead(fc_fd, &fc_len, sizeof(fc_len));

		/* 我们是否到达文件的末尾？ */
		if (fc_nbytes == 0)
			break;

		/* 我们的长度是否正确？ */
		if (fc_nbytes != sizeof(fc_len))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from streaming transaction's changes file \"%s\": read only %zu of %zu bytes",
							fc_path, fc_nbytes, sizeof(fc_len))));

		if (fc_len <= 0)
			elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
				 fc_len, fc_path);

		/* 确保我们有足够大的缓冲区 */
		fc_buffer = repalloc(fc_buffer, fc_len);

		/* 最后将数据读入缓冲区 */
		fc_nbytes = BufFileRead(fc_fd, fc_buffer, fc_len);
		if (fc_nbytes != fc_len)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from streaming transaction's changes file \"%s\": read only %zu of %zu bytes",
							fc_path, fc_nbytes, (size_t) fc_len)));

		/* 将缓冲区复制到stringinfo，并调用apply_dispatch */
		resetStringInfo(&fc_s2);
		appendBinaryStringInfo(&fc_s2, fc_buffer, fc_len);

		/* 确保我们将数据读取到我们的内存上下文中。 */
		fc_oldcxt = MemoryContextSwitchTo(ApplyMessageContext);

		fc_apply_dispatch(&fc_s2);

		MemoryContextReset(ApplyMessageContext);

		MemoryContextSwitchTo(fc_oldcxt);

		fc_nchanges++;

		if (fc_nchanges % 1000 == 0)
			elog(DEBUG1, "replayed %d changes from file \"%s\"",
				 fc_nchanges, fc_path);
	}

	BufFileClose(fc_fd);

	pfree(fc_buffer);
	pfree(fc_s2.data);

	elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
		 fc_nchanges, fc_path);

	return;
}

/*
 * 处理STREAM COMMIT消息。
 */
static void fc_apply_handle_stream_commit(StringInfo fc_s)
{
	TransactionId fc_xid;
	LogicalRepCommitData fc_commit_data;

	if (in_streamed_transaction)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("STREAM COMMIT message without STREAM STOP")));

	fc_xid = logicalrep_read_stream_commit(fc_s, &fc_commit_data);
	fc_set_apply_error_context_xact(fc_xid, fc_commit_data.commit_lsn);

	elog(DEBUG1, "received commit for streamed transaction %u", fc_xid);

	fc_apply_spooled_messages(fc_xid, fc_commit_data.commit_lsn);

	fc_apply_handle_commit_internal(&fc_commit_data);

	/* 取消链接具有序列化更改和子事务信息的文件 */
	fc_stream_cleanup_files(MyLogicalRepWorker->subid, fc_xid);

	/* 处理任何正在并行同步的表。 */
	process_syncing_tables(fc_commit_data.end_lsn);

	pgstat_report_activity(STATE_IDLE, NULL);

	fc_reset_apply_error_context_info();
}

/*
 * apply_handle_commit和apply_handle_stream_commit的辅助函数。
 */
static void fc_apply_handle_commit_internal(LogicalRepCommitData *fc_commit_data)
{
	if (is_skipping_changes())
	{
		fc_stop_skipping_changes();

		/*
		 * 启动一个新的事务以清除subskiplsn，如果尚未开始。
		 */
		if (!IsTransactionState())
			StartTransactionCommand();
	}

	if (IsTransactionState())
	{
		/*
		 * 事务要么是非空的，要么被跳过，因此我们清除subskiplsn。
		 */
		fc_clear_subscription_skip_lsn(fc_commit_data->commit_lsn);

		/*
		 * 更新原始状态，以便在崩溃的情况下可以从正确的位置重新启动流媒体。
		 */
		replorigin_session_origin_lsn = fc_commit_data->end_lsn;
		replorigin_session_origin_timestamp = fc_commit_data->committime;

		CommitTransactionCommand();
		pgstat_report_stat(false);

		fc_store_flush_position(fc_commit_data->end_lsn);
	}
	else
	{
		/* 处理可能累计的任何失效消息。 */
		AcceptInvalidationMessages();
		fc_maybe_reread_subscription();
	}

	in_remote_transaction = false;
}

/*
 * 处理RELATION消息。
 *
 * 请注意，我们在这里不对本地模式进行验证。对本地模式的验证
 * 被推迟到给定关系的首个更改到来时，因为我们只关心在应用更改时
 * 进行验证，并且这样可以减少锁定。
 */
static void fc_apply_handle_relation(StringInfo fc_s)
{
	LogicalRepRelation *fc_rel;

	if (fc_handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, fc_s))
		return;

	fc_rel = logicalrep_read_rel(fc_s);
	logicalrep_relmap_update(fc_rel);

	/* 还要重置所有引用remoterel的分区映射中的条目。 */
	logicalrep_partmap_reset_relmap(fc_rel);
}

/*
 * 处理TYPE消息。
 *
 * 此实现不关注TYPE消息；我们期望用户设置好
 * 使得传入数据对于本地订阅表的输入函数是可接受的。因此，
 * 我们只是读取并丢弃该消息。
 */
static void fc_apply_handle_type(StringInfo fc_s)
{
	LogicalRepTyp fc_typ;

	if (fc_handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, fc_s))
		return;

	logicalrep_read_typ(fc_s, &fc_typ);
}

/*
 * 获取副本身份索引，如果未定义则获取主键。
 *
 * 如果两者都未定义，则返回InvalidOid。
 */
static Oid fc_GetRelationIdentityOrPK(Relation fc_rel)
{
	Oid			fc_idxoid;

	fc_idxoid = RelationGetReplicaIndex(fc_rel);

	if (!OidIsValid(fc_idxoid))
		fc_idxoid = RelationGetPrimaryKeyIndex(fc_rel);

	return fc_idxoid;
}

/*
 * 检查我们（订阅者所有者）是否对目标关系具有执行给定操作的
 * 足够权限。
 */
static void fc_TargetPrivilegesCheck(Relation fc_rel, AclMode fc_mode)
{
	Oid			fc_relid;
	AclResult	fc_aclresult;

	fc_relid = RelationGetRelid(fc_rel);
	fc_aclresult = pg_class_aclcheck(fc_relid, GetUserId(), fc_mode);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult,
					   get_relkind_objtype(fc_rel->rd_rel->relkind),
					   get_rel_name(fc_relid));

	/*
	 * 我们缺乏尊重RLS策略的基础设施。可能在这里添加这样的
	 * 基础设施是可能的，但tablesync工作者也缺乏这项功能，
	 * 所以我们不去麻烦。RLS通常不适用于TRUNCATE命令，
	 * 但看起来复制一个TRUNCATE并拒绝复制后续的INSERT是危险的，
	 * 因此我们禁止所有相同的命令。
	 */
	if (check_enable_rls(fc_relid, InvalidOid, false) == RLS_ENABLED)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
						GetUserNameFromId(GetUserId(), true),
						RelationGetRelationName(fc_rel))));
}

/*
 * 处理INSERT消息。
 */

static void fc_apply_handle_insert(StringInfo fc_s)
{
	LogicalRepRelMapEntry *fc_rel;
	LogicalRepTupleData fc_newtup;
	LogicalRepRelId fc_relid;
	ApplyExecutionData *fc_edata;
	EState	   *fc_estate;
	TupleTableSlot *fc_remoteslot;
	MemoryContext fc_oldctx;
#ifdef FDDM
	bool fc_skip_for_column_missing = false;
#endif
	/*
	 * 如果我们跳过数据修改更改或处理流式事务，则快速返回。
	 */
	if (is_skipping_changes() ||
		fc_handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, fc_s))
		return;

#ifdef FDDM
	if (fdd_skip_transaction)
	{
		ereport(DEBUG1, (errmsg("insert:current transaction has been setted to skip by conflict handler.")));
		return;
	}
#endif
	fc_begin_replication_step();

	fc_relid = logicalrep_read_insert(fc_s, &fc_newtup);
#ifdef FDDM
	if (is_multi_rep_sub && table_missing_handle_hook)
	{
		fc_rel = fdd_logicalrep_rel_open(fc_relid, RowExclusiveLock,&fc_newtup, &fc_skip_for_column_missing, fdd_column_missing_handle_hook);
		if(!fc_rel->localrelvalid)
		{
			//没报错，但是又返回无效，肯定需要mmr插件处理
			bool fc_not_skip = true;

			if (fc_skip_for_column_missing)
			{
				fc_end_replication_step();
				return;
			}

			fc_not_skip = table_missing_handle_hook(&fc_rel->remoterel, &fc_newtup);
			if (!fc_not_skip)
			{
				//跳过
				fc_end_replication_step();
				return;
			}
			ereport(ERROR,
						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						 errmsg("logical replication target relation \"%s.%s\" does not exist",
								fc_rel->remoterel.nspname, fc_rel->remoterel.relname)));
		}
	}
	else
	{
		fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
	}
#else
	fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
#endif
	if (!fc_should_apply_changes_for_rel(fc_rel))
	{
		/*
		 * 关系在事务中间不能变得有趣，因此解锁它是安全的。
		 */
		logicalrep_rel_close(fc_rel, RowExclusiveLock);
		fc_end_replication_step();
		return;
	}

	/* 设置错误回调的关系 */
	apply_error_callback_arg.rel = fc_rel;

	/* 初始化执行者状态。 */
	fc_edata = fc_create_edata_for_relation(fc_rel);
	fc_estate = fc_edata->estate;
	fc_remoteslot = ExecInitExtraTupleSlot(fc_estate,
										RelationGetDescr(fc_rel->localrel),
										&TTSOpsVirtual);

	/* 在槽中处理和存储远程元组 */
	fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
	fc_slot_store_data(fc_remoteslot, fc_rel, &fc_newtup);
	fc_slot_fill_defaults(fc_rel, fc_estate, fc_remoteslot);
	MemoryContextSwitchTo(fc_oldctx);

	/* 对于分区表，将元组插入到一个分区中。 */
	if (fc_rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		fc_apply_handle_tuple_routing(fc_edata,
								   fc_remoteslot, NULL, CMD_INSERT);
	else
		fc_apply_handle_insert_internal(fc_edata, fc_edata->targetRelInfo,
									 fc_remoteslot);

	fc_finish_edata(fc_edata);

	/* 重置错误回调的关系 */
	apply_error_callback_arg.rel = NULL;

	logicalrep_rel_close(fc_rel, NoLock);

	fc_end_replication_step();
}

/*
 * 应用于 apply_handle_insert() 的工作马
 * relinfo 是我们实际插入的关系
 * （可能是 edata->targetRelInfo 的子分区）
 */
static void fc_apply_handle_insert_internal(ApplyExecutionData *fc_edata,
							 ResultRelInfo *fc_relinfo,
							 TupleTableSlot *fc_remoteslot)
{
	EState	   *fc_estate = fc_edata->estate;

	/* 我们必须在这里打开索引。 */
	ExecOpenIndices(fc_relinfo, false);

#ifdef FDDM
	if (is_multi_rep_sub && conflict_handle_hook != NULL)
	{
		//冲突处理
		//bool handled = false;
		ApplyExecState *fc_apestate = palloc0(sizeof(ApplyExecState));
		fc_apestate->estate = fc_estate;
		fc_apestate->epqstate = NULL;
		fc_apestate->resultRelInfo = fc_relinfo;
		fc_apestate->slot = fc_remoteslot;
		fc_apestate->targetRel = fc_edata->targetRel;
		
		conflict_handle_hook(Fdd_Insert, fc_apestate);
		pfree(fc_apestate);
		ExecCloseIndices(fc_relinfo);
		return;
	}

#endif

	/* 执行插入。 */
	fc_TargetPrivilegesCheck(fc_relinfo->ri_RelationDesc, ACL_INSERT);
	ExecSimpleRelationInsert(fc_relinfo, fc_estate, fc_remoteslot);

	/* 清理。 */
	ExecCloseIndices(fc_relinfo);
}

/*
 * 检查逻辑复制关系是否可以更新，如果不能则抛出
 * 适当的错误。
 */
static void fc_check_relation_updatable(LogicalRepRelMapEntry *fc_rel)
{
	/*
	 * 对于分区表，我们只需关心目标分区是否是
	 * 可更新的（即是否为其定义了 PK 或 RI）。
	 */
	if (fc_rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		return;

	/* 可更新，无错误。 */
	if (fc_rel->updatable)
		return;

	/*
	 * 我们处于错误模式，因此这有些慢也没关系。给用户正确的错误
	 * 更好。
	 */
	if (OidIsValid(fc_GetRelationIdentityOrPK(fc_rel->localrel)))
	{
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("publisher did not send replica identity column "
						"expected by the logical replication target relation \"%s.%s\"",
						fc_rel->remoterel.nspname, fc_rel->remoterel.relname)));
	}

	ereport(ERROR,
			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
			 errmsg("logical replication target relation \"%s.%s\" has "
					"neither REPLICA IDENTITY index nor PRIMARY "
					"KEY and published relation does not have "
					"REPLICA IDENTITY FULL",
					fc_rel->remoterel.nspname, fc_rel->remoterel.relname)));
}

/*
 * 处理 UPDATE 消息。
 *
 * TODO: FDW 支持
 */
static void fc_apply_handle_update(StringInfo fc_s)
{
	LogicalRepRelMapEntry *fc_rel;
	LogicalRepRelId fc_relid;
	ApplyExecutionData *fc_edata;
	EState	   *fc_estate;
	LogicalRepTupleData fc_oldtup;
	LogicalRepTupleData fc_newtup;
	bool		fc_has_oldtup;
	TupleTableSlot *fc_remoteslot;
	RangeTblEntry *fc_target_rte;
	MemoryContext fc_oldctx;
#ifdef FDDM
	bool fc_skip_for_column_missing = false;
#endif
	/*
	 * 如果我们跳过数据修改更改或处理流式事务，则快速返回。
	 */
	if (is_skipping_changes() ||
		fc_handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, fc_s))
		return;
	
#ifdef FDDM
	if (fdd_skip_transaction)
	{
		ereport(DEBUG1, (errmsg("update:current transaction has been setted to skip by conflict handler.")));
		return;
	}
#endif

	fc_begin_replication_step();

	fc_relid = logicalrep_read_update(fc_s, &fc_has_oldtup, &fc_oldtup,
								   &fc_newtup);
#ifdef FDDM
	if (is_multi_rep_sub && table_missing_handle_hook)
	{
		fc_rel = fdd_logicalrep_rel_open(fc_relid, RowExclusiveLock,&fc_newtup, &fc_skip_for_column_missing, fdd_column_missing_handle_hook);
		if(!fc_rel->localrelvalid)
		{
			//没报错，但是又返回无效，肯定需要mmr插件处理
			bool fc_not_skip = true;

			if (fc_skip_for_column_missing)
			{
				fc_end_replication_step();
				return;
			}

			fc_not_skip = table_missing_handle_hook(&fc_rel->remoterel, &fc_newtup);
			if (!fc_not_skip)
			{
				//跳过
				fc_end_replication_step();
				return;
			}
			ereport(ERROR,
						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						 errmsg("logical replication target relation \"%s.%s\" does not exist",
								fc_rel->remoterel.nspname, fc_rel->remoterel.relname)));
		}
	}
	else
	{
		fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
	}
#else
	fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
#endif

	if (!fc_should_apply_changes_for_rel(fc_rel))
	{
		/*
		 * 关系在事务中间不能变得有趣，因此解锁它是安全的。
		 */
		logicalrep_rel_close(fc_rel, RowExclusiveLock);
		fc_end_replication_step();
		return;
	}

	/* 设置错误回调的关系 */
	apply_error_callback_arg.rel = fc_rel;

	/* 检查我们是否可以执行更新。 */
	fc_check_relation_updatable(fc_rel);

	/* 初始化执行者状态。 */
	fc_edata = fc_create_edata_for_relation(fc_rel);
	fc_estate = fc_edata->estate;
	fc_remoteslot = ExecInitExtraTupleSlot(fc_estate,
										RelationGetDescr(fc_rel->localrel),
										&TTSOpsVirtual);

	/*
	 * 填充 updatedCols，以便每列触发器可以触发，从而
	 * 执行器可以正确地传递 indexUnchanged 提示。这可能
	 * 包括比在发布者上实际更改的列更多的列，
	 * 因为逻辑复制协议不包含该信息。但是它将
	 * 例如排除仅在订阅者上存在的列，因为我们不接触那些。
	 */
	fc_target_rte = list_nth(fc_estate->es_range_table, 0);
	for (int fc_i = 0; fc_i < fc_remoteslot->tts_tupleDescriptor->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_remoteslot->tts_tupleDescriptor, fc_i);
		int			fc_remoteattnum = fc_rel->attrmap->attnums[fc_i];

		if (!fc_att->attisdropped && fc_remoteattnum >= 0)
		{
			Assert(fc_remoteattnum < fc_newtup.ncols);
			if (fc_newtup.colstatus[fc_remoteattnum] != LOGICALREP_COLUMN_UNCHANGED)
				fc_target_rte->updatedCols =
					bms_add_member(fc_target_rte->updatedCols,
								   fc_i + 1 - FirstLowInvalidHeapAttributeNumber);
		}
	}

	/* 构建搜索元组。 */
	fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
	fc_slot_store_data(fc_remoteslot, fc_rel,
					fc_has_oldtup ? &fc_oldtup : &fc_newtup);
	MemoryContextSwitchTo(fc_oldctx);
#ifdef FDDM
	if (is_multi_rep_sub && conflict_handle_hook != NULL)
	{
		rep_upadte_has_old = fc_has_oldtup;
	}
#endif
	/* 对于分区表，将更新应用于正确的分区。 */
	if (fc_rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		fc_apply_handle_tuple_routing(fc_edata,
								   fc_remoteslot, &fc_newtup, CMD_UPDATE);
	else
		fc_apply_handle_update_internal(fc_edata, fc_edata->targetRelInfo,
									 fc_remoteslot, &fc_newtup);

	fc_finish_edata(fc_edata);

	/* 重置错误回调的关系 */
	apply_error_callback_arg.rel = NULL;

	logicalrep_rel_close(fc_rel, NoLock);

	fc_end_replication_step();
}

/*
 * 应用于 apply_handle_update() 的工作马
 * relinfo 是我们实际更新的关系
 * （可能是 edata->targetRelInfo 的子分区）
 */
static void fc_apply_handle_update_internal(ApplyExecutionData *fc_edata,
							 ResultRelInfo *fc_relinfo,
							 TupleTableSlot *fc_remoteslot,
							 LogicalRepTupleData *fc_newtup)
{
	EState	   *fc_estate = fc_edata->estate;
	LogicalRepRelMapEntry *fc_relmapentry = fc_edata->targetRel;
	Relation	fc_localrel = fc_relinfo->ri_RelationDesc;
	EPQState	fc_epqstate;
	TupleTableSlot *fc_localslot;
	bool		fc_found;
	MemoryContext fc_oldctx;

	EvalPlanQualInit(&fc_epqstate, fc_estate, NULL, NIL, -1);
	ExecOpenIndices(fc_relinfo, false);
#ifdef FDDM
	if (is_multi_rep_sub && conflict_handle_hook != NULL)
	{
		//冲突处理
		//bool handled = false;
		ApplyExecState *fc_apestate = palloc0(sizeof(ApplyExecState));
		fc_apestate->estate = fc_estate;
		fc_apestate->epqstate = &fc_epqstate;
		fc_apestate->resultRelInfo = fc_relinfo;
		fc_apestate->slot = fc_remoteslot;
		fc_apestate->targetRel = fc_edata->targetRel;
		fc_apestate->newtup = fc_newtup;

		conflict_handle_hook(Fdd_Update, fc_apestate);
		pfree(fc_apestate);
		rep_upadte_has_old = false;

		//回调处理完了，退出
		/* 清理。 */
		ExecCloseIndices(fc_relinfo);
		EvalPlanQualEnd(&fc_epqstate);
		return;
	}
#endif

	fc_found = fc_FindReplTupleInLocalRel(fc_estate, fc_localrel,
									&fc_relmapentry->remoterel,
									fc_remoteslot, &fc_localslot);
	ExecClearTuple(fc_remoteslot);

	/*
	 * 找到元组。
	 *
	 * 请注意，如果有其他冲突的唯一索引，这将失败。
	 */
	if (fc_found)
	{
		/* 在槽中处理和存储远程元组 */
		fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
		fc_slot_modify_data(fc_remoteslot, fc_localslot, fc_relmapentry, fc_newtup);
		MemoryContextSwitchTo(fc_oldctx);

		EvalPlanQualSetSlot(&fc_epqstate, fc_remoteslot);

		/* 执行实际更新。 */
		fc_TargetPrivilegesCheck(fc_relinfo->ri_RelationDesc, ACL_UPDATE);
		ExecSimpleRelationUpdate(fc_relinfo, fc_estate, &fc_epqstate, fc_localslot,
								 fc_remoteslot);
	}
	else
	{
		/*
		 * 要更新的元组找不到。除了
		 * 记录日志消息外不做其他操作。
		 *
		 * XXX 这应该被提升为 ereport(LOG) 吗？
		 */
		elog(DEBUG1,
			 "logical replication did not find row to be updated "
			 "in replication target relation \"%s\"",
			 RelationGetRelationName(fc_localrel));
	}

	/* 清理。 */
	ExecCloseIndices(fc_relinfo);
	EvalPlanQualEnd(&fc_epqstate);
}

/*
 * 处理 DELETE 消息。
 *
 * TODO: FDW 支持
 */
static void fc_apply_handle_delete(StringInfo fc_s)
{
	LogicalRepRelMapEntry *fc_rel;
	LogicalRepTupleData fc_oldtup;
	LogicalRepRelId fc_relid;
	ApplyExecutionData *fc_edata;
	EState	   *fc_estate;
	TupleTableSlot *fc_remoteslot;
	MemoryContext fc_oldctx;
#ifdef FDDM
	bool fc_skip_for_column_missing = false;
#endif
	/*
	 * 如果我们跳过数据修改更改或处理流式事务，则快速返回。
	 */
	if (is_skipping_changes() ||
		fc_handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, fc_s))
		return;
		
#ifdef FDDM
	if (fdd_skip_transaction)
	{
		ereport(DEBUG1, (errmsg("delete:current transaction has been setted to skip by conflict handler.")));
		return;
	}
#endif

	fc_begin_replication_step();

	fc_relid = logicalrep_read_delete(fc_s, &fc_oldtup);
#ifdef FDDM
	if (is_multi_rep_sub && table_missing_handle_hook)
	{
		fc_rel = fdd_logicalrep_rel_open(fc_relid, RowExclusiveLock,&fc_oldtup, &fc_skip_for_column_missing, fdd_column_missing_handle_hook);
		if(!fc_rel->localrelvalid)
		{
			//没报错，但是又返回无效，肯定需要mmr插件处理
			bool fc_not_skip = true;

			if (fc_skip_for_column_missing)
			{
				fc_end_replication_step();
				return;
			}

			fc_not_skip = table_missing_handle_hook(&fc_rel->remoterel, &fc_oldtup);
			if (!fc_not_skip)
			{
				//跳过
				fc_end_replication_step();
				return;
			}
			ereport(ERROR,
						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						 errmsg("logical replication target relation \"%s.%s\" does not exist",
								fc_rel->remoterel.nspname, fc_rel->remoterel.relname)));
		}
	}
	else
	{
		fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
	}	
#else
	fc_rel = logicalrep_rel_open(fc_relid, RowExclusiveLock);
#endif
	if (!fc_should_apply_changes_for_rel(fc_rel))
	{
		/*
		 * 关系在事务中间不能变得有趣，因此解锁它是安全的。
		 */
		logicalrep_rel_close(fc_rel, RowExclusiveLock);
		fc_end_replication_step();
		return;
	}

	/* 设置错误回调的关系 */
	apply_error_callback_arg.rel = fc_rel;

	/* 检查我们是否可以执行删除。 */
	fc_check_relation_updatable(fc_rel);

	/* 初始化执行者状态。 */
	fc_edata = fc_create_edata_for_relation(fc_rel);
	fc_estate = fc_edata->estate;
	fc_remoteslot = ExecInitExtraTupleSlot(fc_estate,
										RelationGetDescr(fc_rel->localrel),
										&TTSOpsVirtual);

	/* 构建搜索元组。 */
	fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
	fc_slot_store_data(fc_remoteslot, fc_rel, &fc_oldtup);
	MemoryContextSwitchTo(fc_oldctx);

	/* 对于分区表，将删除应用于正确的分区。 */
	if (fc_rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		fc_apply_handle_tuple_routing(fc_edata,
								   fc_remoteslot, NULL, CMD_DELETE);
	else
		fc_apply_handle_delete_internal(fc_edata, fc_edata->targetRelInfo,
									 fc_remoteslot);

	fc_finish_edata(fc_edata);

	/* 重置错误回调的关系 */
	apply_error_callback_arg.rel = NULL;

	logicalrep_rel_close(fc_rel, NoLock);

	fc_end_replication_step();
}

/*
 * 应用于 apply_handle_delete() 的工作马
 * relinfo 是我们实际删除的关系
 * （可能是 edata->targetRelInfo 的子分区）
 */
static void fc_apply_handle_delete_internal(ApplyExecutionData *fc_edata,
							 ResultRelInfo *fc_relinfo,
							 TupleTableSlot *fc_remoteslot)
{
	EState	   *fc_estate = fc_edata->estate;
	Relation	fc_localrel = fc_relinfo->ri_RelationDesc;
	LogicalRepRelation *fc_remoterel = &fc_edata->targetRel->remoterel;
	EPQState	fc_epqstate;
	TupleTableSlot *fc_localslot;
	bool		fc_found;

	EvalPlanQualInit(&fc_epqstate, fc_estate, NULL, NIL, -1);
	ExecOpenIndices(fc_relinfo, false);
#ifdef FDDM
	if (is_multi_rep_sub && conflict_handle_hook != NULL)
	{
		//冲突处理
		//bool handled = false;
		ApplyExecState *fc_apestate = palloc0(sizeof(ApplyExecState));
		fc_apestate->estate = fc_estate;
		fc_apestate->epqstate = &fc_epqstate;
		fc_apestate->resultRelInfo = fc_relinfo;
		fc_apestate->slot = fc_remoteslot;
		fc_apestate->targetRel = fc_edata->targetRel;
		
		conflict_handle_hook(Fdd_Delete, fc_apestate);

		pfree(fc_apestate);
		/* 清理。 */
		ExecCloseIndices(fc_relinfo);
		EvalPlanQualEnd(&fc_epqstate);
		return;
	}
#endif

	fc_found = fc_FindReplTupleInLocalRel(fc_estate, fc_localrel, fc_remoterel,
									fc_remoteslot, &fc_localslot);

	/* 如果找到则删除。 */
	if (fc_found)
	{
		EvalPlanQualSetSlot(&fc_epqstate, fc_localslot);

		/* 执行实际删除。 */
		fc_TargetPrivilegesCheck(fc_relinfo->ri_RelationDesc, ACL_DELETE);
		ExecSimpleRelationDelete(fc_relinfo, fc_estate, &fc_epqstate, fc_localslot);
	}
	else
	{
		/*
		 * 找不到要删除的元组。 除了发出日志消息外，不执行任何操作。
		 *
		 * XXX 这是否应该提升为 ereport(LOG)？
		 */
		elog(DEBUG1,
			 "logical replication did not find row to be deleted "
			 "in replication target relation \"%s\"",
			 RelationGetRelationName(fc_localrel));
	}

	/* 清理。 */
	ExecCloseIndices(fc_relinfo);
	EvalPlanQualEnd(&fc_epqstate);
}

/*
 * 尝试在对应的本地关系中使用复制身份索引、主键或必要时，顺序扫描
 * 从发布端接收的元组（在 'remoteslot' 中）查找。
 *
 * 如果找到本地元组，将其返回到 '*localslot'。
 */
static bool fc_FindReplTupleInLocalRel(EState *fc_estate, Relation fc_localrel,
						LogicalRepRelation *fc_remoterel,
						TupleTableSlot *fc_remoteslot,
						TupleTableSlot **fc_localslot)
{
	Oid			fc_idxoid;
	bool		fc_found;

	/*
	 * 无论顶层操作是什么，我们在这里执行读取，所以
	 * 检查 SELECT 权限。
	 */
	fc_TargetPrivilegesCheck(fc_localrel, ACL_SELECT);

	*fc_localslot = table_slot_create(fc_localrel, &fc_estate->es_tupleTable);

	fc_idxoid = fc_GetRelationIdentityOrPK(fc_localrel);
	Assert(OidIsValid(fc_idxoid) ||
		   (fc_remoterel->replident == REPLICA_IDENTITY_FULL));

	if (OidIsValid(fc_idxoid))
		fc_found = RelationFindReplTupleByIndex(fc_localrel, fc_idxoid,
											 LockTupleExclusive,
											 fc_remoteslot, *fc_localslot);
	else
		fc_found = RelationFindReplTupleSeq(fc_localrel, LockTupleExclusive,
										 fc_remoteslot, *fc_localslot);

	return fc_found;
}

/*
 * 这处理分区表上的插入、更新、删除。
 */
static void fc_apply_handle_tuple_routing(ApplyExecutionData *fc_edata,
						   TupleTableSlot *fc_remoteslot,
						   LogicalRepTupleData *fc_newtup,
						   CmdType fc_operation)
{
	EState	   *fc_estate = fc_edata->estate;
	LogicalRepRelMapEntry *fc_relmapentry = fc_edata->targetRel;
	ResultRelInfo *fc_relinfo = fc_edata->targetRelInfo;
	Relation	fc_parentrel = fc_relinfo->ri_RelationDesc;
	ModifyTableState *fc_mtstate;
	PartitionTupleRouting *fc_proute;
	ResultRelInfo *fc_partrelinfo;
	Relation	fc_partrel;
	TupleTableSlot *fc_remoteslot_part;
	TupleConversionMap *fc_map;
	MemoryContext fc_oldctx;
	LogicalRepRelMapEntry *fc_part_entry = NULL;
	AttrMap    *fc_attrmap = NULL;

	/* ModifyTableState 对 ExecFindPartition() 是必要的。 */
	fc_edata->mtstate = fc_mtstate = makeNode(ModifyTableState);
	fc_mtstate->ps.plan = NULL;
	fc_mtstate->ps.state = fc_estate;
	fc_mtstate->operation = fc_operation;
	fc_mtstate->resultRelInfo = fc_relinfo;

	/* ... PartitionTupleRouting 也是。 */
	fc_edata->proute = fc_proute = ExecSetupPartitionTupleRouting(fc_estate, fc_parentrel);

	/*
	 * 找到“搜索元组”所属的分区。
	 */
	Assert(fc_remoteslot != NULL);
	fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
#ifdef FDDM
	if (is_multi_rep_sub && fdd_partition_handle_hook != NULL)
	{
		fc_partrelinfo = fdd_partition_handle_hook(fc_mtstate, fc_relinfo, fc_proute,
									fc_remoteslot, fc_estate, &fc_relmapentry->remoterel, fc_newtup);

		if (NULL == fc_partrelinfo)
		{
			//找不到分区表，跳过
			return;
		}
	}
	else
	{
		fc_partrelinfo = ExecFindPartition(fc_mtstate, fc_relinfo, fc_proute,
									fc_remoteslot, fc_estate);
	}
#else
	fc_partrelinfo = ExecFindPartition(fc_mtstate, fc_relinfo, fc_proute,
									fc_remoteslot, fc_estate);
#endif
	Assert(fc_partrelinfo != NULL);
	fc_partrel = fc_partrelinfo->ri_RelationDesc;

	/*
	 * 检查支持的 relkind。 我们需要这个，因为分区可能是
	 * 不支持的 relkind；而且分区集可能会变化，因此在
	 * CREATE/ALTER SUBSCRIPTION 时检查是不够的。
	 */
	CheckSubscriptionRelkind(fc_partrel->rd_rel->relkind,
							 get_namespace_name(RelationGetNamespace(fc_partrel)),
							 RelationGetRelationName(fc_partrel));

	/*
	 * 为了执行以下任何操作，元组必须与
	 * 分区的行类型匹配。 如有必要进行转换或仅复制，
	 * 无论如何使用专用槽来存储元组。
	 */
	fc_remoteslot_part = fc_partrelinfo->ri_PartitionTupleSlot;
	if (fc_remoteslot_part == NULL)
		fc_remoteslot_part = table_slot_create(fc_partrel, &fc_estate->es_tupleTable);
	fc_map = fc_partrelinfo->ri_RootToPartitionMap;
	if (fc_map != NULL)
	{
		fc_attrmap = fc_map->attrMap;
		fc_remoteslot_part = execute_attr_map_slot(fc_attrmap, fc_remoteslot,
												fc_remoteslot_part);
	}
	else
	{
		fc_remoteslot_part = ExecCopySlot(fc_remoteslot_part, fc_remoteslot);
		slot_getallattrs(fc_remoteslot_part);
	}
	MemoryContextSwitchTo(fc_oldctx);

	/* 检查我们是否可以在叶分区上执行更新或删除。 */
	if (fc_operation == CMD_UPDATE || fc_operation == CMD_DELETE)
	{
		fc_part_entry = logicalrep_partition_open(fc_relmapentry, fc_partrel,
											   fc_attrmap);
		fc_check_relation_updatable(fc_part_entry);
	}

	switch (fc_operation)
	{
		case CMD_INSERT:
			fc_apply_handle_insert_internal(fc_edata, fc_partrelinfo,
										 fc_remoteslot_part);
			break;

		case CMD_DELETE:
			fc_apply_handle_delete_internal(fc_edata, fc_partrelinfo,
										 fc_remoteslot_part);
			break;

		case CMD_UPDATE:

			/*
			 * 对于 UPDATE，根据更新的元组是否
			 * 满足分区的约束，执行简单的更新
			 * 分区或将更新的元组移入不同的
			 * 合适分区。
			 */
			{
				TupleTableSlot *fc_localslot;
				ResultRelInfo *fc_partrelinfo_new;
				Relation	fc_partrel_new;
				bool		fc_found;
#ifdef FDDM
				if (is_multi_rep_sub && conflict_handle_hook != NULL)
				{
					ApplyExecState *fc_apestate = palloc0(sizeof(ApplyExecState));
					fc_apestate->estate = fc_estate;
					fc_apestate->resultRelInfo = fc_relinfo;
					fc_apestate->slot = fc_remoteslot;
					fc_apestate->targetRel = fc_edata->targetRel;
					fc_apestate->newtup = fc_newtup;


					fc_apestate->partrelinfo = fc_partrelinfo;
					fc_apestate->edata = fc_edata;
					fc_apestate->remoteslot_part = fc_remoteslot_part;
					fc_apestate->map = fc_map;

					conflict_handle_hook(Fdd_Update_Part, fc_apestate);

					break;
				}
#endif

				/* 从分区中获取匹配的本地元组。 */
				fc_found = fc_FindReplTupleInLocalRel(fc_estate, fc_partrel,
												&fc_part_entry->remoterel,
												fc_remoteslot_part, &fc_localslot);
				if (!fc_found)
				{
					/*
					 * 找不到要更新的元组。 除了发出日志消息外，
					 * 不执行任何操作。
					 *
					 * XXX 这是否应该提升为 ereport(LOG)？
					 */
					elog(DEBUG1,
						 "logical replication did not find row to be updated "
						 "in replication target relation's partition \"%s\"",
						 RelationGetRelationName(fc_partrel));
					return;
				}

				/*
				 * 将更新应用于本地元组，将结果放入
				 * remoteslot_part。
				 */
				fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
				fc_slot_modify_data(fc_remoteslot_part, fc_localslot, fc_part_entry,
								 fc_newtup);
				MemoryContextSwitchTo(fc_oldctx);

				/*
				 * 更新的元组仍然满足当前
				 * 分区的约束吗？
				 */
				if (!fc_partrel->rd_rel->relispartition ||
					ExecPartitionCheck(fc_partrelinfo, fc_remoteslot_part, fc_estate,
									   false))
				{
					/*
					 * 是的，因此简单地更新分区。 我们不在这里调用
					 * apply_handle_update_internal()，因为
					 * 那通常会执行以下工作，以避免重复一些
					 * 之前在分区中查找本地元组时已完成的工作。
					 */
					EPQState	fc_epqstate;

					EvalPlanQualInit(&fc_epqstate, fc_estate, NULL, NIL, -1);
					ExecOpenIndices(fc_partrelinfo, false);

					EvalPlanQualSetSlot(&fc_epqstate, fc_remoteslot_part);
					fc_TargetPrivilegesCheck(fc_partrelinfo->ri_RelationDesc,
										  ACL_UPDATE);
					ExecSimpleRelationUpdate(fc_partrelinfo, fc_estate, &fc_epqstate,
											 fc_localslot, fc_remoteslot_part);
					ExecCloseIndices(fc_partrelinfo);
					EvalPlanQualEnd(&fc_epqstate);
				}
				else
				{
					/* 将元组移动到新分区。 */

					/*
					 * 新分区将使用元组路由找到，
					 * 这只能通过父表进行。 我们可能需要
					 * 将元组转换为父的行类型。 请注意，
					 * 这是在分区中找到的元组，而不是
					 * 此函数接收到的原始搜索元组。
					 */
					if (fc_map)
					{
						TupleConversionMap *fc_PartitionToRootMap =
						convert_tuples_by_name(RelationGetDescr(fc_partrel),
											   RelationGetDescr(fc_parentrel));

						fc_remoteslot =
							execute_attr_map_slot(fc_PartitionToRootMap->attrMap,
												  fc_remoteslot_part, fc_remoteslot);
					}
					else
					{
						fc_remoteslot = ExecCopySlot(fc_remoteslot, fc_remoteslot_part);
						slot_getallattrs(fc_remoteslot);
					}

					/* 找到新分区。 */
					fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
					fc_partrelinfo_new = ExecFindPartition(fc_mtstate, fc_relinfo,
														fc_proute, fc_remoteslot,
														fc_estate);
					MemoryContextSwitchTo(fc_oldctx);
					Assert(fc_partrelinfo_new != fc_partrelinfo);
					fc_partrel_new = fc_partrelinfo_new->ri_RelationDesc;

					/* 检查新分区是否也具有支持的 relkind。 */
					CheckSubscriptionRelkind(fc_partrel_new->rd_rel->relkind,
											 get_namespace_name(RelationGetNamespace(fc_partrel_new)),
											 RelationGetRelationName(fc_partrel_new));

					/* 删除在旧分区中找到的旧元组。 */
					fc_apply_handle_delete_internal(fc_edata, fc_partrelinfo,
												 fc_localslot);

					/* 将新元组插入到新分区中。 */

					/*
					 * 将替代元组转换为匹配目标
					 * 分区行类型。
					 */
					fc_oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(fc_estate));
					fc_remoteslot_part = fc_partrelinfo_new->ri_PartitionTupleSlot;
					if (fc_remoteslot_part == NULL)
						fc_remoteslot_part = table_slot_create(fc_partrel_new,
															&fc_estate->es_tupleTable);
					fc_map = fc_partrelinfo_new->ri_RootToPartitionMap;
					if (fc_map != NULL)
					{
						fc_remoteslot_part = execute_attr_map_slot(fc_map->attrMap,
																fc_remoteslot,
																fc_remoteslot_part);
					}
					else
					{
						fc_remoteslot_part = ExecCopySlot(fc_remoteslot_part,
													   fc_remoteslot);
						slot_getallattrs(fc_remoteslot);
					}
					MemoryContextSwitchTo(fc_oldctx);
					fc_apply_handle_insert_internal(fc_edata, fc_partrelinfo_new,
												 fc_remoteslot_part);
				}
			}
			break;

		default:
			elog(ERROR, "unrecognized CmdType: %d", (int) fc_operation);
			break;
	}
}

/*
 * 处理 TRUNCATE 消息。
 *
 * TODO: FDW 支持
 */
static void fc_apply_handle_truncate(StringInfo fc_s)
{
	bool		fc_cascade = false;
	bool		fc_restart_seqs = false;
	List	   *fc_remote_relids = NIL;
	List	   *fc_remote_rels = NIL;
	List	   *fc_rels = NIL;
	List	   *fc_part_rels = NIL;
	List	   *fc_relids = NIL;
	List	   *fc_relids_logged = NIL;
	ListCell   *fc_lc;
	LOCKMODE	fc_lockmode = AccessExclusiveLock;

	/*
	 * 如果我们跳过数据修改更改或处理流式事务，则快速返回。
	 */
	if (is_skipping_changes() ||
		fc_handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, fc_s))
		return;

	fc_begin_replication_step();

	fc_remote_relids = logicalrep_read_truncate(fc_s, &fc_cascade, &fc_restart_seqs);

	foreach(fc_lc, fc_remote_relids)
	{
		LogicalRepRelId fc_relid = lfirst_oid(fc_lc);
		LogicalRepRelMapEntry *fc_rel;

		fc_rel = logicalrep_rel_open(fc_relid, fc_lockmode);
		if (!fc_should_apply_changes_for_rel(fc_rel))
		{
			/*
			 * 关系在事务中间不能变得有趣，因此解锁它是安全的。
			 */
			logicalrep_rel_close(fc_rel, fc_lockmode);
			continue;
		}

		fc_remote_rels = lappend(fc_remote_rels, fc_rel);
		fc_TargetPrivilegesCheck(fc_rel->localrel, ACL_TRUNCATE);
		fc_rels = lappend(fc_rels, fc_rel->localrel);
		fc_relids = lappend_oid(fc_relids, fc_rel->localreloid);
		if (RelationIsLogicallyLogged(fc_rel->localrel))
			fc_relids_logged = lappend_oid(fc_relids_logged, fc_rel->localreloid);

		/*
		 * 如果收到截断分区表的消息，则截断分区。
		 */
		if (fc_rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		{
			ListCell   *fc_child;
			List	   *fc_children = find_all_inheritors(fc_rel->localreloid,
													   fc_lockmode,
													   NULL);

			foreach(fc_child, fc_children)
			{
				Oid			fc_childrelid = lfirst_oid(fc_child);
				Relation	fc_childrel;

				if (list_member_oid(fc_relids, fc_childrelid))
					continue;

				/* find_all_inheritors 已经获得锁 */
				fc_childrel = table_open(fc_childrelid, NoLock);

				/*
				 * 忽略其他后端的临时表。请参见
				 * ExecuteTruncate() 中的类似代码。
				 */
				if (RELATION_IS_OTHER_TEMP(fc_childrel))
				{
					table_close(fc_childrel, fc_lockmode);
					continue;
				}

				fc_TargetPrivilegesCheck(fc_childrel, ACL_TRUNCATE);
				fc_rels = lappend(fc_rels, fc_childrel);
				fc_part_rels = lappend(fc_part_rels, fc_childrel);
				fc_relids = lappend_oid(fc_relids, fc_childrelid);
				/* 仅在逻辑解码所需时记录此关系 */
				if (RelationIsLogicallyLogged(fc_childrel))
					fc_relids_logged = lappend_oid(fc_relids_logged, fc_childrelid);
			}
		}
	}

	/*
	 * 即使我们在上游主节点上使用了 CASCADE，我们仍然明确默认为
	 * 重放更改而不进一步级联。这可能稍后可以使用用户指定的选项进行更改。
	 */
	ExecuteTruncateGuts(fc_rels,
						fc_relids,
						fc_relids_logged,
						DROP_RESTRICT,
						fc_restart_seqs);
	foreach(fc_lc, fc_remote_rels)
	{
		LogicalRepRelMapEntry *fc_rel = lfirst(fc_lc);

		logicalrep_rel_close(fc_rel, NoLock);
	}
	foreach(fc_lc, fc_part_rels)
	{
		Relation	fc_rel = lfirst(fc_lc);

		table_close(fc_rel, NoLock);
	}

	fc_end_replication_step();
}


/*
 * 逻辑复制协议消息调度器。
 */
static void fc_apply_dispatch(StringInfo fc_s)
{
	LogicalRepMsgType fc_action = pq_getmsgbyte(fc_s);
	LogicalRepMsgType fc_saved_command;

	/*
	 * 设置当前正在应用的命令。由于此函数可以在应用已排队的更改时递归调用，
	 * 因此保存当前命令。
	 */
	fc_saved_command = apply_error_callback_arg.command;
	apply_error_callback_arg.command = fc_action;

	switch (fc_action)
	{
		case LOGICAL_REP_MSG_BEGIN:
			fc_apply_handle_begin(fc_s);
			break;

		case LOGICAL_REP_MSG_COMMIT:
			fc_apply_handle_commit(fc_s);
			break;

		case LOGICAL_REP_MSG_INSERT:
			fc_apply_handle_insert(fc_s);
			break;

		case LOGICAL_REP_MSG_UPDATE:
			fc_apply_handle_update(fc_s);
			break;

		case LOGICAL_REP_MSG_DELETE:
			fc_apply_handle_delete(fc_s);
			break;

		case LOGICAL_REP_MSG_TRUNCATE:
			fc_apply_handle_truncate(fc_s);
			break;

		case LOGICAL_REP_MSG_RELATION:
			fc_apply_handle_relation(fc_s);
			break;

		case LOGICAL_REP_MSG_TYPE:
			fc_apply_handle_type(fc_s);
			break;

		case LOGICAL_REP_MSG_ORIGIN:
			fc_apply_handle_origin(fc_s);
			break;

		case LOGICAL_REP_MSG_MESSAGE:

			/*
			 * 逻辑复制仍未使用通用逻辑消息。
			 * 不过，它可以被使用此输出插件的其他应用程序使用。
			 */
			break;

		case LOGICAL_REP_MSG_STREAM_START:
			fc_apply_handle_stream_start(fc_s);
			break;

		case LOGICAL_REP_MSG_STREAM_STOP:
			fc_apply_handle_stream_stop(fc_s);
			break;

		case LOGICAL_REP_MSG_STREAM_ABORT:
			fc_apply_handle_stream_abort(fc_s);
			break;

		case LOGICAL_REP_MSG_STREAM_COMMIT:
			fc_apply_handle_stream_commit(fc_s);
			break;

		case LOGICAL_REP_MSG_BEGIN_PREPARE:
			fc_apply_handle_begin_prepare(fc_s);
			break;

		case LOGICAL_REP_MSG_PREPARE:
			fc_apply_handle_prepare(fc_s);
			break;

		case LOGICAL_REP_MSG_COMMIT_PREPARED:
			fc_apply_handle_commit_prepared(fc_s);
			break;

		case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
			fc_apply_handle_rollback_prepared(fc_s);
			break;

		case LOGICAL_REP_MSG_STREAM_PREPARE:
			fc_apply_handle_stream_prepare(fc_s);
			break;

		default:
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("invalid logical replication message type \"??? (%d)\"", fc_action)));
	}

	/* 重置当前命令 */
	apply_error_callback_arg.command = fc_saved_command;
}

/*
 * 确定要报告给 walsender 进程的写入/刷新位置。
 *
 * 我们不能简单地报告 walsender 发给我们的最后 LSN，因为
 * 本地事务可能尚未在本地刷新到磁盘。相反，我们
 * 构建一个将本地 LSN 与远程 LSN 关联的列表，以便于每次提交。
 * 当向发送者报告刷新位置时，我们遍历该列表，并
 * 检查其中哪些条目已在本地刷新。我们可以将这些报告为已刷新的。
 *
 * 如果存在需要刷新的未结束事务，则 have_pending_txes 为真。
 */
static void fc_get_flush_position(XLogRecPtr *fc_write, XLogRecPtr *fc_flush,
				   bool *fc_have_pending_txes)
{
	dlist_mutable_iter fc_iter;
	XLogRecPtr	fc_local_flush = GetFlushRecPtr(NULL);

	*fc_write = InvalidXLogRecPtr;
	*fc_flush = InvalidXLogRecPtr;

	dlist_foreach_modify(fc_iter, &lsn_mapping)
	{
		FlushPosition *fc_pos =
		dlist_container(FlushPosition, node, fc_iter.cur);

		*fc_write = fc_pos->remote_end;

		if (fc_pos->local_end <= fc_local_flush)
		{
			*fc_flush = fc_pos->remote_end;
			dlist_delete(fc_iter.cur);
			pfree(fc_pos);
		}
		else
		{
			/*
			 * 不想对剩余列表进行无用的迭代，
			 * 该列表可能会很长。相反，获取最后一个元素并
			 * 从中获取写入位置。
			 */
			fc_pos = dlist_tail_element(FlushPosition, node,
									 &lsn_mapping);
			*fc_write = fc_pos->remote_end;
			*fc_have_pending_txes = true;
			return;
		}
	}

	*fc_have_pending_txes = !dlist_is_empty(&lsn_mapping);
}

/*
 * 在跟踪列表中存储当前远程/本地 lsn 对。
 */
static void fc_store_flush_position(XLogRecPtr fc_remote_lsn)
{
	FlushPosition *fc_flushpos;

	/* 需要在永久上下文中执行此操作 */
	MemoryContextSwitchTo(ApplyContext);

	/* 跟踪提交 lsn  */
	fc_flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
	fc_flushpos->local_end = XactLastCommitEnd;
	fc_flushpos->remote_end = fc_remote_lsn;

	dlist_push_tail(&lsn_mapping, &fc_flushpos->node);
	MemoryContextSwitchTo(ApplyMessageContext);
}


/* 更新工作进程的统计信息。 */
static void fc_UpdateWorkerStats(XLogRecPtr fc_last_lsn, TimestampTz fc_send_time, bool fc_reply)
{
	MyLogicalRepWorker->last_lsn = fc_last_lsn;
	MyLogicalRepWorker->last_send_time = fc_send_time;
	MyLogicalRepWorker->last_recv_time = GetCurrentTimestamp();
	if (fc_reply)
	{
		MyLogicalRepWorker->reply_lsn = fc_last_lsn;
		MyLogicalRepWorker->reply_time = fc_send_time;
	}
}

#ifdef FDDM
//检查需要转发的订阅端是否改变，如果改变了，需要重启
static void FDD_CheckForwardSubs(void)
{
	//这里要读取改变的成员变量不是本进程独有的
	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
	if (MyLogicalRepWorker->forward_subs_change)
	{
		MyLogicalRepWorker->forward_subs_change = false;
		LWLockRelease(LogicalRepWorkerLock);
		//自动重启
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will restart because of forward subs changed",
						MySubscription->name)));

		proc_exit(0);
	}
	
	LWLockRelease(LogicalRepWorkerLock);
}
#endif

/*
 * 应用主循环。
 */
static void fc_LogicalRepApplyLoop(XLogRecPtr fc_last_received)
{
	TimestampTz fc_last_recv_timestamp = GetCurrentTimestamp();
	bool		fc_ping_sent = false;
	TimeLineID	fc_tli;
	ErrorContextCallback fc_errcallback;

	/*
	 * 初始化 ApplyMessageContext，我们在每个复制
	 * 协议消息后进行清理。
	 */
	ApplyMessageContext = AllocSetContextCreate(ApplyContext,
												"ApplyMessageContext",
												ALLOCSET_DEFAULT_SIZES);

	/*
	 * 此内存上下文用于启用流模式时的每个流数据。
	 * 每次流停止时，该上下文会被重置。
	 */
	LogicalStreamingContext = AllocSetContextCreate(ApplyContext,
													"LogicalStreamingContext",
													ALLOCSET_DEFAULT_SIZES);

	/* 标记为空闲状态，然后开始循环 */
	pgstat_report_activity(STATE_IDLE, NULL);

	/*
	 * 推送应用错误上下文回调。字段将在应用
	 * 更改时填写。
	 */
	fc_errcallback.callback = fc_apply_error_callback;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 该外层循环每次等待迭代一次。 */
	for (;;)
	{
		pgsocket	fc_fd = PGINVALID_SOCKET;
		int			fc_rc;
		int			fc_len;
		char	   *fc_buf = NULL;
		bool		fc_endofstream = false;
		long		fc_wait_time;

		CHECK_FOR_INTERRUPTS();

		MemoryContextSwitchTo(ApplyMessageContext);

		fc_len = walrcv_receive(LogRepWorkerWalRcvConn, &fc_buf, &fc_fd);

		if (fc_len != 0)
		{
			/* 循环处理所有可用数据（不阻塞）。 */
			for (;;)
			{
				CHECK_FOR_INTERRUPTS();

				if (fc_len == 0)
				{
					break;
				}
				else if (fc_len < 0)
				{
					ereport(LOG,
							(errmsg("data stream from publisher has ended")));
					fc_endofstream = true;
					break;
				}
				else
				{
					int			fc_c;
					StringInfoData fc_s;

					/* 重置超时。 */
					fc_last_recv_timestamp = GetCurrentTimestamp();
					fc_ping_sent = false;

					/* 确保我们将数据读取到我们的内存上下文中。 */
					MemoryContextSwitchTo(ApplyMessageContext);

					fc_s.data = fc_buf;
					fc_s.len = fc_len;
					fc_s.cursor = 0;
					fc_s.maxlen = -1;

					fc_c = pq_getmsgbyte(&fc_s);

					if (fc_c == 'w')
					{
						XLogRecPtr	fc_start_lsn;
						XLogRecPtr	fc_end_lsn;
						TimestampTz fc_send_time;

						fc_start_lsn = pq_getmsgint64(&fc_s);
						fc_end_lsn = pq_getmsgint64(&fc_s);
						fc_send_time = pq_getmsgint64(&fc_s);

						if (fc_last_received < fc_start_lsn)
							fc_last_received = fc_start_lsn;

						if (fc_last_received < fc_end_lsn)
							fc_last_received = fc_end_lsn;

						fc_UpdateWorkerStats(fc_last_received, fc_send_time, false);

						fc_apply_dispatch(&fc_s);
					}
					else if (fc_c == 'k')
					{
						XLogRecPtr	fc_end_lsn;
						TimestampTz fc_timestamp;
						bool		fc_reply_requested;

						fc_end_lsn = pq_getmsgint64(&fc_s);
						fc_timestamp = pq_getmsgint64(&fc_s);
						fc_reply_requested = pq_getmsgbyte(&fc_s);

						if (fc_last_received < fc_end_lsn)
							fc_last_received = fc_end_lsn;

						fc_send_feedback(fc_last_received, fc_reply_requested, false);
						fc_UpdateWorkerStats(fc_last_received, fc_timestamp, true);
					}
					/* 其他消息类型被故意忽略 */

					MemoryContextReset(ApplyMessageContext);
				}

				fc_len = walrcv_receive(LogRepWorkerWalRcvConn, &fc_buf, &fc_fd);
			}
		}

		/* 确认到目前为止的所有写入 */
		fc_send_feedback(fc_last_received, false, false);

		if (!in_remote_transaction && !in_streamed_transaction)
		{
			/*
			 * 如果我们一段时间没有收到任何事务，队列中可能会有
			 * 未处理的无效消息，现在处理它们。
			 */
			AcceptInvalidationMessages();
			fc_maybe_reread_subscription();

			/* 处理任何表同步的变化。 */
			process_syncing_tables(fc_last_received);
		}

		/* 清理内存。 */
		MemoryContextResetAndDeleteChildren(ApplyMessageContext);
		MemoryContextSwitchTo(TopMemoryContext);

		/* 检查我们是否需要退出流传输循环。 */
		if (fc_endofstream)
			break;

		/*
		 * 等待更多数据或锁存。如果我们有未刷新事务，
		 * 在 WalWriterDelay 之后唤醒，以查看它们是否已被刷新（在
		 * 这种情况下我们应该发送反馈消息）。否则，除非我们收到数据或信号，
		 * 否则没有特别的紧急性需要唤醒。
		 */
		if (!dlist_is_empty(&lsn_mapping))
			fc_wait_time = WalWriterDelay;
		else
			fc_wait_time = NAPTIME_PER_CYCLE;

		fc_rc = WaitLatchOrSocket(MyLatch,
							   WL_SOCKET_READABLE | WL_LATCH_SET |
							   WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
							   fc_fd, fc_wait_time,
							   WAIT_EVENT_LOGICAL_APPLY_MAIN);

		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
#ifdef FDDM
			FDD_CheckForwardSubs();
#endif
			CHECK_FOR_INTERRUPTS();
		}

		if (ConfigReloadPending)
		{
			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);
		}

		if (fc_rc & WL_TIMEOUT)
		{
			/*
			 * 我们没有收到任何新消息。如果我们在 wal_receiver_timeout / 2
			 * 之前没有收到来自服务器的任何消息，向服务器发送 ping。
			 * 此外，如果自上次我们发送的更新以来已经超过
			 * wal_receiver_status_interval，仍然向主服务器发送状态更新，
			 * 以报告应用 WAL 的任何进展。
			 */
			bool		fc_requestReply = false;

			/*
			 * 检查自上次从主服务器接收数据以来的时间是否已达到
			 * 配置的限制。
			 */
			if (wal_receiver_timeout > 0)
			{
				TimestampTz fc_now = GetCurrentTimestamp();
				TimestampTz fc_timeout;

				fc_timeout =
					TimestampTzPlusMilliseconds(fc_last_recv_timestamp,
												wal_receiver_timeout);

				if (fc_now >= fc_timeout)
					ereport(ERROR,
							(errcode(ERRCODE_CONNECTION_FAILURE),
							 errmsg("terminating logical replication worker due to timeout")));

				/* 检查是否是时候发送 ping 了。 */
				if (!fc_ping_sent)
				{
					fc_timeout = TimestampTzPlusMilliseconds(fc_last_recv_timestamp,
														  (wal_receiver_timeout / 2));
					if (fc_now >= fc_timeout)
					{
						fc_requestReply = true;
						fc_ping_sent = true;
					}
				}
			}

			fc_send_feedback(fc_last_received, fc_requestReply, fc_requestReply);

			/*
			 * 强制报告以确保长时间的空闲期不会导致
			 * 随意的延迟统计。统计只能在（隐式或显式）事务外报告。
			 * 这不应该导致统计被延迟很长时间，因为事务要么
			 * 在提交时作为一个整体发送，要么进行流式处理。
			 * 流式处理的事务被溢出到磁盘并在提交时应用。
			 */
			if (!IsTransactionState())
				pgstat_report_stat(true);
		}
	}

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;

	/* 完成 */
	walrcv_endstreaming(LogRepWorkerWalRcvConn, &fc_tli);
}

/*
 * 向服务器发送一个备用状态更新消息。
 *
 * 'recvpos' 是我们接收到数据的最新 LSN，如果需要
 * 发送响应以避免超时，则设置强制。
 */
static void fc_send_feedback(XLogRecPtr fc_recvpos, bool fc_force, bool fc_requestReply)
{
	static StringInfo fc_reply_message = NULL;
	static TimestampTz fc_send_time = 0;

	static XLogRecPtr fc_last_recvpos = InvalidXLogRecPtr;
	static XLogRecPtr fc_last_writepos = InvalidXLogRecPtr;
	static XLogRecPtr fc_last_flushpos = InvalidXLogRecPtr;

	XLogRecPtr	fc_writepos;
	XLogRecPtr	fc_flushpos;
	TimestampTz fc_now;
	bool		fc_have_pending_txes;

	/*
	 * 如果用户不希望向发布者报告状态，请确保在执行任何操作之前退出。
	 */
	if (!fc_force && wal_receiver_status_interval <= 0)
		return;

	/* 不传递 recvpos 是合法的 */
	if (fc_recvpos < fc_last_recvpos)
		fc_recvpos = fc_last_recvpos;

	fc_get_flush_position(&fc_writepos, &fc_flushpos, &fc_have_pending_txes);

	/*
	 * 没有未清除的事务，我们可以报告最新接收的位置。
	 * 这对于同步复制很重要。
	 */
	if (!fc_have_pending_txes)
		fc_flushpos = fc_writepos = fc_recvpos;

	if (fc_writepos < fc_last_writepos)
		fc_writepos = fc_last_writepos;

	if (fc_flushpos < fc_last_flushpos)
		fc_flushpos = fc_last_flushpos;

	fc_now = GetCurrentTimestamp();

	/* 如果我们已经报告了所有内容，那么我们很好 */
	if (!fc_force &&
		fc_writepos == fc_last_writepos &&
		fc_flushpos == fc_last_flushpos &&
		!TimestampDifferenceExceeds(fc_send_time, fc_now,
									wal_receiver_status_interval * 1000))
		return;
	fc_send_time = fc_now;

	if (!fc_reply_message)
	{
		MemoryContext fc_oldctx = MemoryContextSwitchTo(ApplyContext);

		fc_reply_message = makeStringInfo();
		MemoryContextSwitchTo(fc_oldctx);
	}
	else
		resetStringInfo(fc_reply_message);

	pq_sendbyte(fc_reply_message, 'r');
	pq_sendint64(fc_reply_message, fc_recvpos);	/* 写入 */
	pq_sendint64(fc_reply_message, fc_flushpos);	/* 刷新 */
	pq_sendint64(fc_reply_message, fc_writepos);	/* 应用 */
	pq_sendint64(fc_reply_message, fc_now);	/* 发送时间 */
	pq_sendbyte(fc_reply_message, fc_requestReply);	/* 请求回复 */

	elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X",
		 fc_force,
		 LSN_FORMAT_ARGS(fc_recvpos),
		 LSN_FORMAT_ARGS(fc_writepos),
		 LSN_FORMAT_ARGS(fc_flushpos));

	walrcv_send(LogRepWorkerWalRcvConn,
				fc_reply_message->data, fc_reply_message->len);

	if (fc_recvpos > fc_last_recvpos)
		fc_last_recvpos = fc_recvpos;
	if (fc_writepos > fc_last_writepos)
		fc_last_writepos = fc_writepos;
	if (fc_flushpos > fc_last_flushpos)
		fc_last_flushpos = fc_flushpos;
}

/*
 * 如有必要，重新读取订阅信息。大多数更改将是退出。
 */
static void fc_maybe_reread_subscription(void)
{
	MemoryContext fc_oldctx;
	Subscription *fc_newsub;
	bool		fc_started_tx = false;

	/* 当缓存状态有效时，这里没有任何事情可做。 */
	if (MySubscriptionValid)
		return;

	/* 此函数可能在事务内或事务外被调用。 */
	if (!IsTransactionState())
	{
		StartTransactionCommand();
		fc_started_tx = true;
	}

	/* 确保在永久上下文中分配。 */
	fc_oldctx = MemoryContextSwitchTo(ApplyContext);

	fc_newsub = GetSubscription(MyLogicalRepWorker->subid, true);

	/*
	 * 如果订阅被移除则退出。这通常不应该发生，
	 * 因为工作线程在 DROP SUBSCRIPTION 期间被杀死。
	 */
	if (!fc_newsub)
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will "
						"stop because the subscription was removed",
						MySubscription->name)));

		proc_exit(0);
	}

	/* 如果订阅被禁用则退出。 */
	if (!fc_newsub->enabled)
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will "
						"stop because the subscription was disabled",
						MySubscription->name)));

		proc_exit(0);
	}

	/* 当启用为真时，!slotname 永远不应该发生。 */
	Assert(fc_newsub->slotname);

	/* 两阶段不应被更改 */
	Assert(fc_newsub->twophasestate == MySubscription->twophasestate);

	/*
	 * 如果任何影响远程连接的参数被更改，则退出。
	 * 启动器将启动一个新的工作进程。
	 */
	if (strcmp(fc_newsub->conninfo, MySubscription->conninfo) != 0 ||
		strcmp(fc_newsub->name, MySubscription->name) != 0 ||
		strcmp(fc_newsub->slotname, MySubscription->slotname) != 0 ||
		fc_newsub->binary != MySubscription->binary ||
		fc_newsub->stream != MySubscription->stream ||
		fc_newsub->owner != MySubscription->owner ||
		!equal(fc_newsub->publications, MySubscription->publications))
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will restart because of a parameter change",
						MySubscription->name)));

		proc_exit(0);
	}

	/* 检查其他不应发生的更改。 */
	if (fc_newsub->dbid != MySubscription->dbid)
	{
		elog(ERROR, "subscription %u changed unexpectedly",
			 MyLogicalRepWorker->subid);
	}

	/* 清理旧的订阅信息并切换到新的信息。 */
	FreeSubscription(MySubscription);
	MySubscription = fc_newsub;

	MemoryContextSwitchTo(fc_oldctx);

	/* 根据用户的愿望更改同步提交 */
	SetConfigOption("synchronous_commit", MySubscription->synccommit,
					PGC_BACKEND, PGC_S_OVERRIDE);

	if (fc_started_tx)
		CommitTransactionCommand();

	MySubscriptionValid = true;
}

/*
 * 订阅系统缓存失效的回调。
 */
static void fc_subscription_change_cb(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue)
{
	MySubscriptionValid = false;
}

/*
 * subxact_info_write
 *	  存储顶层事务的子事务信息。
 *
 * 对于每个子事务，我们在主文件中存储其第一次更改的偏移量。
 * 文件始终被整体覆盖。
 *
 * XXX 我们应该只存储尚未中止的子事务。
 */
static void fc_subxact_info_write(Oid fc_subid, TransactionId fc_xid)
{
	char		fc_path[MAXPGPATH];
	Size		fc_len;
	BufFile    *fc_fd;

	Assert(TransactionIdIsValid(fc_xid));

	/* 构建子事务文件名 */
	fc_subxact_filename(fc_path, fc_subid, fc_xid);

	/* 删除子事务文件（如果存在）。 */
	if (subxact_data.nsubxacts == 0)
	{
		fc_cleanup_subxact_info();
		BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, fc_path, true);

		return;
	}

	/*
	 * 如果子事务文件尚未创建，则创建子事务文件，否则打开
	 * 现有文件。
	 */
	fc_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, fc_path, O_RDWR,
							true);
	if (fc_fd == NULL)
		fc_fd = BufFileCreateFileSet(MyLogicalRepWorker->stream_fileset, fc_path);

	fc_len = sizeof(SubXactInfo) * subxact_data.nsubxacts;

	/* 写入子事务计数和子事务信息 */
	BufFileWrite(fc_fd, &subxact_data.nsubxacts, sizeof(subxact_data.nsubxacts));
	BufFileWrite(fc_fd, subxact_data.subxacts, fc_len);

	BufFileClose(fc_fd);

	/* 释放为子事务信息分配的内存 */
	fc_cleanup_subxact_info();
}

/*
 * subxact_info_read
 *	  恢复有关流式事务的子事务信息。
 *
 * 将关于子事务的信息读取到结构 subxact_data 中，以便后续使用。
 */
static void fc_subxact_info_read(Oid fc_subid, TransactionId fc_xid)
{
	char		fc_path[MAXPGPATH];
	size_t		fc_nread;
	Size		fc_len;
	BufFile    *fc_fd;
	MemoryContext fc_oldctx;

	Assert(!subxact_data.subxacts);
	Assert(subxact_data.nsubxacts == 0);
	Assert(subxact_data.nsubxacts_max == 0);

	/*
	 * 如果子事务文件不存在，这意味着我们没有任何子事务
	 * 信息。
	 */
	fc_subxact_filename(fc_path, fc_subid, fc_xid);
	fc_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, fc_path, O_RDONLY,
							true);
	if (fc_fd == NULL)
		return;

	/* 读取子事务项的数量 */
	fc_nread = BufFileRead(fc_fd, &subxact_data.nsubxacts, sizeof(subxact_data.nsubxacts));
	if (fc_nread != sizeof(subxact_data.nsubxacts))
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read from streaming transaction's subxact file \"%s\": read only %zu of %zu bytes",
						fc_path, fc_nread, sizeof(subxact_data.nsubxacts))));

	fc_len = sizeof(SubXactInfo) * subxact_data.nsubxacts;

	/* 我们将最大值保持为2的幂 */
	subxact_data.nsubxacts_max = 1 << my_log2(subxact_data.nsubxacts);

	/*
	 * 在逻辑流上下文中分配子事务信息。我们在整个流期间需要
	 * 这些信息，以便可以将子事务信息添加到其中。流停止时，我们
	 * 将把这些信息刷新到子事务文件，并重置逻辑流上下文。
	 */
	fc_oldctx = MemoryContextSwitchTo(LogicalStreamingContext);
	subxact_data.subxacts = palloc(subxact_data.nsubxacts_max *
								   sizeof(SubXactInfo));
	MemoryContextSwitchTo(fc_oldctx);

	if (fc_len > 0)
	{
		fc_nread = BufFileRead(fc_fd, subxact_data.subxacts, fc_len);
		if (fc_nread != fc_len)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read from streaming transaction's subxact file \"%s\": read only %zu of %zu bytes",
						fc_path, fc_nread, fc_len)));
	}

	BufFileClose(fc_fd);
}

/*
 * subxact_info_add
 *	  添加有关子事务的信息（主文件中的偏移量）。
 */
static void fc_subxact_info_add(TransactionId fc_xid)
{
	SubXactInfo *fc_subxacts = subxact_data.subxacts;
	int64		fc_i;

	/* 我们必须有一个有效的顶级流 xid 和一个流 fd。 */
	Assert(TransactionIdIsValid(stream_xid));
	Assert(stream_fd != NULL);

	/*
	 * 如果 XID 与顶层事务匹配，我们不想添加它。
	 */
	if (stream_xid == fc_xid)
		return;

	/*
	 * 在大多数情况下，我们正在检查上一个调用中已经看到的同一
	 * 子事务，因此确保忽略它（此更改稍后会出现）。
	 */
	if (subxact_data.subxact_last == fc_xid)
		return;

	/* 好吧，记住我们正在处理这个 XID。 */
	subxact_data.subxact_last = fc_xid;

	/*
	 * 检查事务是否已经存在于子事务的数组中。我们故意从
	 * 尾部扫描数组，因为我们很可能是在为最近的子事务添加
	 * 更改。
	 *
	 * XXX 我们能依赖子事务 XID 按排序顺序到达吗？这将
	 * 允许我们在这里使用二分查找。
	 */
	for (fc_i = subxact_data.nsubxacts; fc_i > 0; fc_i--)
	{
		/* 找到，因此我们完成了 */
		if (fc_subxacts[fc_i - 1].xid == fc_xid)
			return;
	}

	/* 这是一个新的子事务，因此我们需要将其添加到数组中。 */
	if (subxact_data.nsubxacts == 0)
	{
		MemoryContext fc_oldctx;

		subxact_data.nsubxacts_max = 128;

		/*
		 * 在每个流上下文中为子事务分配此内存，请参见
		 * subxact_info_read。
		 */
		fc_oldctx = MemoryContextSwitchTo(LogicalStreamingContext);
		fc_subxacts = palloc(subxact_data.nsubxacts_max * sizeof(SubXactInfo));
		MemoryContextSwitchTo(fc_oldctx);
	}
	else if (subxact_data.nsubxacts == subxact_data.nsubxacts_max)
	{
		subxact_data.nsubxacts_max *= 2;
		fc_subxacts = repalloc(fc_subxacts,
							subxact_data.nsubxacts_max * sizeof(SubXactInfo));
	}

	fc_subxacts[subxact_data.nsubxacts].xid = fc_xid;

	/*
	 * 获取流文件的当前偏移量，并将其存储为
	 * 此子事务的偏移量。
	 */
	BufFileTell(stream_fd,
				&fc_subxacts[subxact_data.nsubxacts].fileno,
				&fc_subxacts[subxact_data.nsubxacts].offset);

	subxact_data.nsubxacts++;
	subxact_data.subxacts = fc_subxacts;
}

/* 格式化包含子事务信息的文件名 */
static inline void fc_subxact_filename(char *fc_path, Oid fc_subid, TransactionId fc_xid)
{
	snprintf(fc_path, MAXPGPATH, "%u-%u.subxacts", fc_subid, fc_xid);
}

/* 格式化包含序列化更改的文件名 */
static inline void fc_changes_filename(char *fc_path, Oid fc_subid, TransactionId fc_xid)
{
	snprintf(fc_path, MAXPGPATH, "%u-%u.changes", fc_subid, fc_xid);
}

/*
 * stream_cleanup_files
 *	  清理订阅 / 顶层事务的文件。
 *
 * 删除与特定顶层事务相关的序列化更改和子事务信息文件。每个订阅都有一组单独的文件
 * 用于任何顶层事务。
 */
static void fc_stream_cleanup_files(Oid fc_subid, TransactionId fc_xid)
{
	char		fc_path[MAXPGPATH];

	/* 删除更改文件。 */
	fc_changes_filename(fc_path, fc_subid, fc_xid);
	BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, fc_path, false);

	/* 如果存在，则删除子事务文件。 */
	fc_subxact_filename(fc_path, fc_subid, fc_xid);
	BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, fc_path, true);
}

/*
 * stream_open_file
 *	  打开一个文件，用于序列化顶层事务的更改。
 *
 * 为由 stream_xid（全局变量）标识的顶层事务的流式更改打开一个文件。如果这是
 * 此事务的第一块流式更改，创建 buffile，否则打开之前创建的文件。
 *
 * 这只能在“流式传输”块的开头调用，即在上游的 stream_start/stream_stop 消息之间。
 */
static void fc_stream_open_file(Oid fc_subid, TransactionId fc_xid, bool fc_first_segment)
{
	char		fc_path[MAXPGPATH];
	MemoryContext fc_oldcxt;

	Assert(in_streamed_transaction);
	Assert(OidIsValid(fc_subid));
	Assert(TransactionIdIsValid(fc_xid));
	Assert(stream_fd == NULL);


	fc_changes_filename(fc_path, fc_subid, fc_xid);
	elog(DEBUG1, "opening file \"%s\" for streamed changes", fc_path);

	/*
	 * 在逻辑流式上下文中创建/打开 buffiles，以便我们
	 * 在流停止之前保留这些文件。
	 */
	fc_oldcxt = MemoryContextSwitchTo(LogicalStreamingContext);

	/*
	 * 如果这是第一段流式数据，创建更改文件。
	 * 否则，只需以附加模式打开文件进行写入。
	 */
	if (fc_first_segment)
		stream_fd = BufFileCreateFileSet(MyLogicalRepWorker->stream_fileset,
										 fc_path);
	else
	{
		/*
		 * 打开文件并将指针移动到文件末尾，因为我们总是
		 * 将更改附加到文件。
		 */
		stream_fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset,
									   fc_path, O_RDWR, false);
		BufFileSeek(stream_fd, 0, 0, SEEK_END);
	}

	MemoryContextSwitchTo(fc_oldcxt);
}

/*
 * stream_close_file
 *	  关闭当前打开的流式更改文件。
 *
 * 这只能在流式块的末尾调用，即在上游的 stream_stop
 * 消息处。
 */
static void fc_stream_close_file(void)
{
	Assert(in_streamed_transaction);
	Assert(TransactionIdIsValid(stream_xid));
	Assert(stream_fd != NULL);

	BufFileClose(stream_fd);

	stream_xid = InvalidTransactionId;
	stream_fd = NULL;
}

/*
 * stream_write_change
 *	  将更改序列化到当前顶层事务的文件中。
 *
 * 更改以简单格式序列化，包括长度（不包括
 * 长度）、操作码（标识消息类型）和消息
 * 内容（不包含子事务的 TransactionId 值）。
 */
static void fc_stream_write_change(char fc_action, StringInfo fc_s)
{
	int			fc_len;

	Assert(in_streamed_transaction);
	Assert(TransactionIdIsValid(stream_xid));
	Assert(stream_fd != NULL);

	/* 磁盘上的总大小，包括操作类型字符 */
	fc_len = (fc_s->len - fc_s->cursor) + sizeof(char);

	/* 首先写入大小 */
	BufFileWrite(stream_fd, &fc_len, sizeof(fc_len));

	/* 然后是操作 */
	BufFileWrite(stream_fd, &fc_action, sizeof(fc_action));

	/* 最后是缓冲区的其余部分（在 XID 之后） */
	fc_len = (fc_s->len - fc_s->cursor);

	BufFileWrite(stream_fd, &fc_s->data[fc_s->cursor], fc_len);
}

/*
 * 清理子事务的内存并重置相关变量。
 */
static inline void fc_cleanup_subxact_info()
{
	if (subxact_data.subxacts)
		pfree(subxact_data.subxacts);

	subxact_data.subxacts = NULL;
	subxact_data.subxact_last = InvalidTransactionId;
	subxact_data.nsubxacts = 0;
	subxact_data.nsubxacts_max = 0;
}

/*
 * 为两阶段事务构造准备好的事务 GID。
 *
 * 在提供的缓冲区中返回 GID。
 */
static void fc_TwoPhaseTransactionGid(Oid fc_subid, TransactionId fc_xid, char *fc_gid, int fc_szgid)
{
	Assert(fc_subid != InvalidRepOriginId);

	if (!TransactionIdIsValid(fc_xid))
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg_internal("invalid two-phase transaction ID")));

	snprintf(fc_gid, fc_szgid, "pg_gid_%u_%u", fc_subid, fc_xid);
}

/*
 * 执行初始同步并处理错误。禁用订阅，
 * 如果需要的话。
 *
 * 在返回时在长生命周期上下文中分配插槽名称。注意，我们不
 * 处理由于系统资源错误引起的致命错误，这些错误是不可重复的。
 */
static void fc_start_table_sync(XLogRecPtr *fc_origin_startpos, char **fc_myslotname)
{
	char	   *fc_syncslotname = NULL;

	Assert(am_tablesync_worker());

	PG_TRY();
	{
		/* 调用初始同步。 */
		fc_syncslotname = LogicalRepSyncTableStart(fc_origin_startpos);
	}
	PG_CATCH();
	{
		if (MySubscription->disableonerr)
			fc_DisableSubscriptionAndExit();
		else
		{
			/*
			 * 报告工作者在表同步过程中失败。中止
			 * 当前事务，以便在空闲状态下发送统计消息。
			 */
			AbortOutOfAnyTransaction();
			pgstat_report_subscription_error(MySubscription->oid, false);

			PG_RE_THROW();
		}
	}
	PG_END_TRY();

	/* 在长生命周期上下文中分配插槽名称 */
	*fc_myslotname = MemoryContextStrdup(ApplyContext, fc_syncslotname);
	pfree(fc_syncslotname);
}

/*
 * 在应用循环中运行并处理错误。禁用订阅，
 * 如有必要。
 *
 * 注意，我们不处理由于
 * 系统资源错误引起的致命错误，这些错误是不可重复的。
 */
static void fc_start_apply(XLogRecPtr fc_origin_startpos)
{
	PG_TRY();
	{
		fc_LogicalRepApplyLoop(fc_origin_startpos);
	}
	PG_CATCH();
	{
		if (MySubscription->disableonerr)
			fc_DisableSubscriptionAndExit();
		else
		{
			/*
			 * 在应用更改时报告工作失败。中止当前事务，以便在空闲状态下发送统计消息。
			 */
			AbortOutOfAnyTransaction();
			pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());

			PG_RE_THROW();
		}
	}
	PG_END_TRY();
}

/* 逻辑复制应用工作线程入口点 */
void ApplyWorkerMain(Datum fc_main_arg)
{
	int			fc_worker_slot = DatumGetInt32(fc_main_arg);
	MemoryContext fc_oldctx;
	char		fc_originname[NAMEDATALEN];
	XLogRecPtr	fc_origin_startpos = InvalidXLogRecPtr;
	char	   *fc_myslotname = NULL;
	WalRcvStreamOptions fc_options;
	int			fc_server_version;

	/* 附加到槽 */
	logicalrep_worker_attach(fc_worker_slot);

	/* 设置信号处理 */
	pqsignal(SIGHUP, SignalHandlerForConfigReload);
	pqsignal(SIGTERM, die);
	BackgroundWorkerUnblockSignals();

	/*
	 * 我们当前在 walreceiver 进程中不需要任何 ResourceOwner，
	 * 但如果需要的话，我们可以在这里调用 CreateAuxProcessResourceOwner。
	 */

	/* 将统计信息初始化为合理的值 */
	MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
		MyLogicalRepWorker->reply_time = GetCurrentTimestamp();

	/* 加载 libpq 特定函数 */
	load_file("libpqwalreceiver", false);

	/* 作为副本会话复制角色运行。 */
	SetConfigOption("session_replication_role", "replica",
					PGC_SUSET, PGC_S_OVERRIDE);

	/* 连接到我们的数据库。 */
	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
											  MyLogicalRepWorker->userid,
											  0);

	/*
	 * 设置始终安全的搜索路径，以便恶意用户无法重定向用户
	 * 代码（例如 pg_index.indexprs）。
	 */
	SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);

	/* 将订阅加载到持久内存上下文中。 */
	ApplyContext = AllocSetContextCreate(TopMemoryContext,
										 "ApplyContext",
										 ALLOCSET_DEFAULT_SIZES);
	StartTransactionCommand();
	fc_oldctx = MemoryContextSwitchTo(ApplyContext);

	MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
	if (!MySubscription)
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription %u will not "
						"start because the subscription was removed during startup",
						MyLogicalRepWorker->subid)));
		proc_exit(0);
	}

#ifdef FDDM
	if (sub_judge_normal_hook)
	{
		is_multi_rep_sub = sub_judge_normal_hook(MyLogicalRepWorker->subid, MySubscription->name);
	}

#endif

	MySubscriptionValid = true;
	MemoryContextSwitchTo(fc_oldctx);

	if (!MySubscription->enabled)
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will not "
						"start because the subscription was disabled during startup",
						MySubscription->name)));

		proc_exit(0);
	}

	/* 根据用户的意愿设置同步提交 */
	SetConfigOption("synchronous_commit", MySubscription->synccommit,
					PGC_BACKEND, PGC_S_OVERRIDE);

	/* 保持我们对订阅更改的了解。 */
	CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
								  fc_subscription_change_cb,
								  (Datum) 0);

	if (am_tablesync_worker())
		ereport(LOG,
				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
						MySubscription->name, get_rel_name(MyLogicalRepWorker->relid))));
	else
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" has started",
						MySubscription->name)));

	CommitTransactionCommand();

	/* 连接到源并开始复制。 */
	elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
		 MySubscription->conninfo);

	if (am_tablesync_worker())
	{
		fc_start_table_sync(&fc_origin_startpos, &fc_myslotname);

		/*
		 * 在长期存活的上下文中分配源名称，以便于错误上下文
		 * 消息。
		 */
		ReplicationOriginNameForTablesync(MySubscription->oid,
										  MyLogicalRepWorker->relid,
										  fc_originname,
										  sizeof(fc_originname));
		apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
																   fc_originname);
	}
	else
	{
		/* 这是主要的应用工作线程 */
		RepOriginId fc_originid;
		TimeLineID	fc_startpointTLI;
		char	   *fc_err;

		fc_myslotname = MySubscription->slotname;

		/*
		 * 如果订阅已启用，这不应该发生，但要防范 DDL 错误或手动目录更改。
		 * （libpqwalreceiver 如果槽为 NULL 将会崩溃。）
		 */
		if (!fc_myslotname)
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("subscription has no replication slot set")));

		/* 设置复制源跟踪。 */
		StartTransactionCommand();
		snprintf(fc_originname, sizeof(fc_originname), "pg_%u", MySubscription->oid);
		fc_originid = replorigin_by_name(fc_originname, true);
		if (!OidIsValid(fc_originid))
			fc_originid = replorigin_create(fc_originname);
		replorigin_session_setup(fc_originid);
		replorigin_session_origin = fc_originid;
		fc_origin_startpos = replorigin_session_get_progress(false);
		CommitTransactionCommand();

		LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
												MySubscription->name, &fc_err);
		if (LogRepWorkerWalRcvConn == NULL)
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not connect to the publisher: %s", fc_err)));

		/*
		 * 我们实际上并不将 output identify_system 用于任何事情，但它
		 * 会在上游进行一些初始化，因此我们仍然调用它。
		 */
		(void) walrcv_identify_system(LogRepWorkerWalRcvConn, &fc_startpointTLI);

		/*
		 * 在长期存活的上下文中分配源名称，以便于错误上下文
		 * 消息。
		 */
		apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
																   fc_originname);
	}

	/*
	 * 为 syscache 设置回调，以便我们知道何时订阅关系状态发生变化。
	 */
	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
								  invalidate_syncing_table_states,
								  (Datum) 0);

	/* 构建逻辑复制流选项。 */
	fc_options.logical = true;
	fc_options.startpoint = fc_origin_startpos;
	fc_options.slotname = fc_myslotname;

	fc_server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
	fc_options.proto.logical.proto_version =
		fc_server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
		fc_server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
		LOGICALREP_PROTO_VERSION_NUM;

	fc_options.proto.logical.publication_names = MySubscription->publications;
	fc_options.proto.logical.binary = MySubscription->binary;
	fc_options.proto.logical.streaming = MySubscription->stream;
	fc_options.proto.logical.twophase = false;

#ifdef FDDM
	//必须是多活订阅端，才去获取forward_subs
	//没有初始化，先初始化
	fc_options.forward_sub_names = NULL;
	if (is_multi_rep_sub && fdd_get_forward_subs_hook != NULL)
	{
		List *fc_subs_names = fdd_get_forward_subs_hook(MyLogicalRepWorker->subid);

		fc_options.forward_sub_names = fc_subs_names;
	}
#endif

	if (!am_tablesync_worker())
	{
		/*
		 * 即使用户请求了 two_phase 模式，它仍然保持为三态 PENDING，
		 * 直到所有表同步达到 READY 状态。只有这样，它才能变为 ENABLED。
		 *
		 * 注意：如果订阅没有表，则将状态保持为 PENDING，
		 * 这允许 ALTER SUBSCRIPTION ... REFRESH PUBLICATION 正常工作。
		 */
		if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
			AllTablesyncsReady())
		{
			/* 启动带有 two_phase 启用的流式传输 */
			fc_options.proto.logical.twophase = true;
			walrcv_startstreaming(LogRepWorkerWalRcvConn, &fc_options);

			StartTransactionCommand();
			UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
			MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
			CommitTransactionCommand();
		}
		else
		{
			walrcv_startstreaming(LogRepWorkerWalRcvConn, &fc_options);
		}

		ereport(DEBUG1,
				(errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
						MySubscription->name,
						MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
						MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
						MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
						"?")));
	}
	else
	{
		/* 启动正常的逻辑流复制。 */
		walrcv_startstreaming(LogRepWorkerWalRcvConn, &fc_options);
	}

	/* 运行主循环。 */
	fc_start_apply(fc_origin_startpos);

	proc_exit(0);
}

/*
 * 在错误恢复后，在新的事务中禁用订阅并正常退出。
 */
static void fc_DisableSubscriptionAndExit(void)
{
	/*
	 * 发出错误消息，并从错误状态恢复到空闲
	 * 状态。
	 */
	HOLD_INTERRUPTS();

	EmitErrorReport();
	AbortOutOfAnyTransaction();
	FlushErrorState();

	RESUME_INTERRUPTS();

	/* 报告工作线程在表同步或应用过程中失败 */
	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
									 !am_tablesync_worker());

	/* 禁用订阅 */
	StartTransactionCommand();
	DisableSubscription(MySubscription->oid);
	CommitTransactionCommand();

	/* 通知订阅已被禁用并退出 */
	ereport(LOG,
			errmsg("subscription \"%s\" has been disabled because of an error",
				   MySubscription->name));

	proc_exit(0);
}

/*
 * 当前进程是逻辑复制工作线程吗？
 */
bool IsLogicalWorker(void)
{
	return MyLogicalRepWorker != NULL;
}

/*
 * 如果给定 LSN 匹配订阅的 skiplsn，则开始跳过事务的更改。
 */
static void fc_maybe_start_skipping_changes(XLogRecPtr fc_finish_lsn)
{
	Assert(!is_skipping_changes());
	Assert(!in_remote_transaction);
	Assert(!in_streamed_transaction);

	/*
	 * 如果没有请求跳过这个事务则快速返回。这个
	 * 函数会被每个远程事务调用，我们假设
	 * 跳过事务的情况不常用。
	 */
	if (likely(XLogRecPtrIsInvalid(MySubscription->skiplsn) ||
			   MySubscription->skiplsn != fc_finish_lsn))
		return;

	/* 开始跳过此事务的所有更改 */
	skip_xact_finish_lsn = fc_finish_lsn;

	ereport(LOG,
			errmsg("logical replication starts skipping transaction at LSN %X/%X",
				   LSN_FORMAT_ARGS(skip_xact_finish_lsn)));
}

/*
 * 通过重置skip_xact_finish_lsn来停止跳过更改（如果启用）。
 */
static void fc_stop_skipping_changes(void)
{
	if (!is_skipping_changes())
		return;

	ereport(LOG,
			(errmsg("logical replication completed skipping transaction at LSN %X/%X",
					LSN_FORMAT_ARGS(skip_xact_finish_lsn))));

	/* 停止跳过更改 */
	skip_xact_finish_lsn = InvalidXLogRecPtr;
}

/*
 * 清除pg_subscription目录的subskiplsn。
 *
 * finish_lsn是事务的完成LSN，用于检查
 * subskiplsn是否与之匹配。如果不匹配，在清除
 * subskiplsn时我们会抛出警告，以提醒用户例如
 * 用户错误地指定了错误的subskiplsn。
 */
static void fc_clear_subscription_skip_lsn(XLogRecPtr fc_finish_lsn)
{
	Relation	fc_rel;
	Form_pg_subscription fc_subform;
	HeapTuple	fc_tup;
	XLogRecPtr	fc_myskiplsn = MySubscription->skiplsn;
	bool		fc_started_tx = false;

	if (likely(XLogRecPtrIsInvalid(fc_myskiplsn)))
		return;

	if (!IsTransactionState())
	{
		StartTransactionCommand();
		fc_started_tx = true;
	}

	/*
	 * 在清除subskiplsn时保护pg_subscription的subskiplsn，防止同时被更新。
	 */
	LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
					 AccessShareLock);

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	/* 获取现有元组。 */
	fc_tup = SearchSysCacheCopy1(SUBSCRIPTIONOID,
							  ObjectIdGetDatum(MySubscription->oid));

	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "subscription \"%s\" does not exist", MySubscription->name);

	fc_subform = (Form_pg_subscription) GETSTRUCT(fc_tup);

	/*
	 * 清除subskiplsn。如果用户在清除之前已经更改了subskiplsn，
	 * 我们不会更新目录，复制源状态将不会前进。所以在最坏的情况下，
	 * 如果服务器在发送冲刷位置的确认之前崩溃，
	 * 该事务将会再次发送，用户需要再次设置subskiplsn。我们可以
	 * 通过记录一个复制源的WAL记录来减少这个可能性，
	 * 以推进源LSN，但没有办法推进
	 * 源时间戳，并且这似乎不值得做，
	 * 因为这是一种非常罕见的情况。
	 */
	if (fc_subform->subskiplsn == fc_myskiplsn)
	{
		bool		fc_nulls[Natts_pg_subscription];
		bool		fc_replaces[Natts_pg_subscription];
		Datum		fc_values[Natts_pg_subscription];

		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, false, sizeof(fc_nulls));
		memset(fc_replaces, false, sizeof(fc_replaces));

		/* 重置subskiplsn */
		fc_values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
		fc_replaces[Anum_pg_subscription_subskiplsn - 1] = true;

		fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_values, fc_nulls,
								fc_replaces);
		CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

		if (fc_myskiplsn != fc_finish_lsn)
			ereport(WARNING,
					errmsg("skip-LSN of subscription \"%s\" cleared", MySubscription->name),
					errdetail("Remote transaction's finish WAL location (LSN) %X/%X did not match skip-LSN %X/%X.",
							  LSN_FORMAT_ARGS(fc_finish_lsn),
							  LSN_FORMAT_ARGS(fc_myskiplsn)));
	}

	heap_freetuple(fc_tup);
	table_close(fc_rel, NoLock);

	if (fc_started_tx)
		CommitTransactionCommand();
}

/* 错误回调以提供关于正在应用的更改的更多上下文信息 */
static void fc_apply_error_callback(void *fc_arg)
{
	ApplyErrorCallbackArg *fc_errarg = &apply_error_callback_arg;

	if (apply_error_callback_arg.command == 0)
		return;

	Assert(fc_errarg->origin_name);

	if (fc_errarg->rel == NULL)
	{
		if (!TransactionIdIsValid(fc_errarg->remote_xid))
			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
					   fc_errarg->origin_name,
					   logicalrep_message_type(fc_errarg->command));
		else if (XLogRecPtrIsInvalid(fc_errarg->finish_lsn))
			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
					   fc_errarg->origin_name,
					   logicalrep_message_type(fc_errarg->command),
					   fc_errarg->remote_xid);
		else
			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
					   fc_errarg->origin_name,
					   logicalrep_message_type(fc_errarg->command),
					   fc_errarg->remote_xid,
					   LSN_FORMAT_ARGS(fc_errarg->finish_lsn));
	}
	else if (fc_errarg->remote_attnum < 0)
		errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%X",
				   fc_errarg->origin_name,
				   logicalrep_message_type(fc_errarg->command),
				   fc_errarg->rel->remoterel.nspname,
				   fc_errarg->rel->remoterel.relname,
				   fc_errarg->remote_xid,
				   LSN_FORMAT_ARGS(fc_errarg->finish_lsn));
	else
		errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u, finished at %X/%X",
				   fc_errarg->origin_name,
				   logicalrep_message_type(fc_errarg->command),
				   fc_errarg->rel->remoterel.nspname,
				   fc_errarg->rel->remoterel.relname,
				   fc_errarg->rel->remoterel.attnames[fc_errarg->remote_attnum],
				   fc_errarg->remote_xid,
				   LSN_FORMAT_ARGS(fc_errarg->finish_lsn));
}

/* 设置应用错误回调的事务信息 */
static inline void fc_set_apply_error_context_xact(TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	apply_error_callback_arg.remote_xid = fc_xid;
	apply_error_callback_arg.finish_lsn = fc_lsn;
}

/* 重置应用错误回调的所有信息 */
static inline void fc_reset_apply_error_context_info(void)
{
	apply_error_callback_arg.command = 0;
	apply_error_callback_arg.rel = NULL;
	apply_error_callback_arg.remote_attnum = -1;
	fc_set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
}
